fix(db): preserve expression-based indexes in autogenerate (#144)

Closes #140. SQLAlchemy 2.0 can't reflect functional indexes (e.g.
`CREATE INDEX ... ON t (lower(email))`) under the SQLite dialect, so
Alembic autogenerate silently skips them — leaving SQLite dev DBs
without an index that production Postgres has.

- Add `make_process_revision_directives(metadata)` in
  `simple_module_db.migrations`: walks each generated `MigrationScript`
  and, for any `CreateTableOp` whose table has expression-based indexes
  in the target metadata, appends a matching `CreateIndexOp` (and the
  reverse `DropIndexOp` in the downgrade). Dedups against already-emitted
  ops so the hook is a no-op on Postgres.
- Wire the hook into `host/migrations/env.py` and the scaffold template
  so every existing and `smpy new`-generated host gets it.
- Add catch-up migration `41cf2c53660e` that back-fills
  `ix_users_user_email_lower` on databases already past
  `3bf3f9db7f7f`. Uses `if_not_exists` / `if_exists` so it's a no-op
  on installs that somehow already have the index.

Author: antosubash

framework/cli/simple_module_cli/templates/host/migrations/env.py

@@ -11,7 +11,12 @@
 from logging.config import fileConfig
 
 from alembic import context
-from simple_module_db import build_module_metadata, make_include_object, render_item
+from simple_module_db import (
+    build_module_metadata,
+    make_include_object,
+    make_process_revision_directives,
+    render_item,
+)
 from simple_module_hosting.settings import Settings
 from sqlalchemy import engine_from_config, pool
 
@@ -24,6 +29,9 @@
 
 target_metadata = build_module_metadata()
 include_object = make_include_object(target_metadata)
+# Re-emit expression-based indexes (e.g. ``lower(email)``) that autogenerate
+# silently drops under SQLite. See ``make_process_revision_directives`` docstring.
+process_revision_directives = make_process_revision_directives(target_metadata)
 
 
 def _get_url() -> str:
@@ -45,6 +53,7 @@ def run_migrations_offline() -> None:
         dialect_opts={"paramstyle": "named"},
         include_object=include_object,
         render_item=render_item,
+        process_revision_directives=process_revision_directives,
     )
 
     with context.begin_transaction():
@@ -68,6 +77,7 @@ def run_migrations_online() -> None:
             target_metadata=target_metadata,
             include_object=include_object,
             render_item=render_item,
+            process_revision_directives=process_revision_directives,
         )
 
         with context.begin_transaction():

framework/db/simple_module_db/__init__.py

@@ -3,7 +3,12 @@
 from simple_module_db.base import create_module_base
 from simple_module_db.deps import get_db
 from simple_module_db.listeners import TenantIsolationError, current_tenant_id
-from simple_module_db.migrations import build_module_metadata, make_include_object, render_item
+from simple_module_db.migrations import (
+    build_module_metadata,
+    make_include_object,
+    make_process_revision_directives,
+    render_item,
+)
 from simple_module_db.mixins import AuditMixin, MultiTenantMixin, SoftDeleteMixin, VersionedMixin
 from simple_module_db.provider import DatabaseProvider, detect_provider
 from simple_module_db.session import DatabaseState, init_db
@@ -23,5 +28,6 @@
     "get_db",
     "init_db",
     "make_include_object",
+    "make_process_revision_directives",
     "render_item",
 ]

framework/db/simple_module_db/migrations.py

@@ -21,9 +21,10 @@
 from typing import Literal
 
 import sqlalchemy as sa
+from alembic.operations.ops import CreateIndexOp, CreateTableOp, DropIndexOp, DropTableOp
 from simple_module_core import ModuleBase
 from simple_module_core.discovery import discover_modules, get_module_package_name
-from sqlalchemy import MetaData
+from sqlalchemy import Column, Index, MetaData
 from sqlalchemy.schema import SchemaItem
 
 from simple_module_db.base import all_module_bases
@@ -36,6 +37,7 @@
     "schema", "table", "column", "index", "unique_constraint", "foreign_key_constraint"
 ]
 IncludeObjectFn = Callable[[SchemaItem, str | None, _SchemaItemType, bool, SchemaItem | None], bool]
+ProcessRevisionDirectivesFn = Callable[[object, object, list], None]
 
 
 def build_module_metadata(modules: Sequence[ModuleBase] | None = None) -> MetaData:
@@ -110,6 +112,97 @@ def include_object(
     return include_object
 
 
+def make_process_revision_directives(
+    metadata: MetaData,
+) -> ProcessRevisionDirectivesFn:
+    """Return an Alembic ``process_revision_directives`` hook that re-adds
+    expression-based indexes silently dropped by autogenerate.
+
+    SQLAlchemy 2.0 can't reflect expression-based indexes (functional indexes
+    like ``CREATE INDEX ... ON t (lower(email))``) under the SQLite dialect.
+    Autogenerate guards against false-positive diffs there by *skipping* the
+    index entirely — which is correct for an "existing table, can't tell if
+    the index already exists" diff, but disastrous on initial CREATE TABLE:
+    SQLite dev DBs end up without an index that production Postgres has.
+
+    This hook walks each generated ``MigrationScript`` and, for every
+    ``CreateTableOp`` whose target table has expression-based indexes in the
+    metadata, appends a matching ``CreateIndexOp``. The reverse ``DropIndexOp``
+    is inserted into ``downgrade_ops`` before the table drop for symmetry —
+    not strictly required (dropping the table drops the index) but it keeps
+    autogen output readable.
+
+    Call as::
+
+        context.configure(
+            ...,
+            process_revision_directives=make_process_revision_directives(target_metadata),
+        )
+    """
+    expression_indexes: dict[str, list[Index]] = {}
+    for table in metadata.tables.values():
+        for index in table.indexes:
+            if _index_is_expression_based(index):
+                expression_indexes.setdefault(table.name, []).append(index)
+
+    def process_revision_directives(context, revision, directives):
+        if not expression_indexes:
+            return
+        for script in directives:
+            upgrade_ops = getattr(script, "upgrade_ops", None)
+            if upgrade_ops is not None:
+                _inject_create_index_after_create_table(upgrade_ops, expression_indexes)
+            downgrade_ops = getattr(script, "downgrade_ops", None)
+            if downgrade_ops is not None:
+                _inject_drop_index_before_drop_table(downgrade_ops, expression_indexes)
+
+    return process_revision_directives
+
+
+def _index_is_expression_based(index: Index) -> bool:
+    """An index is expression-based when any of its expressions is not a plain ``Column``."""
+    return any(not isinstance(expr, Column) for expr in index.expressions)
+
+
+def _inject_create_index_after_create_table(upgrade_ops, expression_indexes) -> None:
+    existing_index_names = {
+        getattr(op, "index_name", None) for op in upgrade_ops.ops if isinstance(op, CreateIndexOp)
+    }
+    new_ops: list = []
+    for op in upgrade_ops.ops:
+        new_ops.append(op)
+        if not isinstance(op, CreateTableOp):
+            continue
+        for index in expression_indexes.get(op.table_name, []):
+            if index.name in existing_index_names:
+                continue
+            new_ops.append(CreateIndexOp.from_index(index))
+            existing_index_names.add(index.name)
+            logger.info(
+                "Re-emitting expression-based index %r on %r — autogenerate "
+                "skipped it (dialect can't reflect functional indexes).",
+                index.name,
+                op.table_name,
+            )
+    upgrade_ops.ops = new_ops
+
+
+def _inject_drop_index_before_drop_table(downgrade_ops, expression_indexes) -> None:
+    existing_drop_names = {
+        getattr(op, "index_name", None) for op in downgrade_ops.ops if isinstance(op, DropIndexOp)
+    }
+    new_ops: list = []
+    for op in downgrade_ops.ops:
+        if isinstance(op, DropTableOp):
+            for index in expression_indexes.get(op.table_name, []):
+                if index.name in existing_drop_names:
+                    continue
+                new_ops.append(DropIndexOp(index.name, table_name=op.table_name))
+                existing_drop_names.add(index.name)
+        new_ops.append(op)
+    downgrade_ops.ops = new_ops
+
+
 def render_item(type_, obj, autogen_context):
     """Alembic ``render_item`` callback for SQLModel + extension types.
 

framework/db/tests/test_migrations.py

@@ -1,4 +1,4 @@
-"""Tests for build_module_metadata and make_include_object (Gap 1)."""
+"""Tests for the simple_module_db.migrations helpers."""
 
 from __future__ import annotations
 
@@ -93,3 +93,110 @@ async def test_include_object_skips_unmodeled_cross_module_fks_by_default(self):
             include(stranger_fk, "fk_stranger", "foreign_key_constraint", False, stranger_fk)
             is False
         )
+
+
+class TestProcessRevisionDirectives:
+    """Autogenerate silently drops expression-based indexes (e.g. ``lower(email)``)
+    on SQLite. ``make_process_revision_directives`` re-injects them when the
+    target table is being newly created in the same revision."""
+
+    def _build_meta(self, *, expression_index: bool = True):
+        from sqlalchemy import Column, Index, Integer, MetaData, String, Table, text
+
+        meta = MetaData()
+        t = Table(
+            "things",
+            meta,
+            Column("id", Integer, primary_key=True),
+            Column("email", String(320)),
+        )
+        if expression_index:
+            Index("ix_things_email_lower", text("lower(email)"), _table=t)
+        else:
+            Index("ix_things_email", t.c.email)
+        return meta
+
+    def _build_directives(self, table_name: str, *extra_ops, empty: bool = False):
+        from alembic.operations.ops import (
+            CreateTableOp,
+            DowngradeOps,
+            DropTableOp,
+            MigrationScript,
+            UpgradeOps,
+        )
+        from sqlalchemy import Column, Integer, String
+
+        if empty:
+            upgrade_ops_list, downgrade_ops_list = [], []
+        else:
+            create_table = CreateTableOp(
+                table_name,
+                [Column("id", Integer, primary_key=True), Column("email", String(320))],
+            )
+            upgrade_ops_list = [create_table, *extra_ops]
+            downgrade_ops_list = [DropTableOp(table_name)]
+        return [
+            MigrationScript(
+                rev_id="abc123",
+                upgrade_ops=UpgradeOps(ops=upgrade_ops_list),
+                downgrade_ops=DowngradeOps(ops=downgrade_ops_list),
+                message="test",
+            )
+        ]
+
+    def test_injects_expression_index_after_create_table(self):
+        """A functional index in metadata is appended as ``CreateIndexOp`` after
+        the matching ``CreateTableOp``, and the reverse drop is inserted before
+        the ``DropTableOp`` in the downgrade."""
+        from alembic.operations.ops import CreateIndexOp, DropIndexOp
+        from simple_module_db.migrations import make_process_revision_directives
+
+        directives = self._build_directives("things")
+        make_process_revision_directives(self._build_meta())(None, None, directives)
+
+        index_ops = [op for op in directives[0].upgrade_ops.ops if isinstance(op, CreateIndexOp)]
+        assert len(index_ops) == 1
+        assert index_ops[0].index_name == "ix_things_email_lower"
+        assert index_ops[0].table_name == "things"
+
+        drop_ops = [op for op in directives[0].downgrade_ops.ops if isinstance(op, DropIndexOp)]
+        assert len(drop_ops) == 1
+        assert drop_ops[0].index_name == "ix_things_email_lower"
+
+    def test_does_not_inject_when_no_create_table(self):
+        """If the revision is not creating the table (e.g. a pure data migration
+        or unrelated change), the hook should not append a CreateIndexOp."""
+        from alembic.operations.ops import CreateIndexOp
+        from simple_module_db.migrations import make_process_revision_directives
+
+        directives = self._build_directives("things", empty=True)
+        make_process_revision_directives(self._build_meta())(None, None, directives)
+
+        assert not any(isinstance(op, CreateIndexOp) for op in directives[0].upgrade_ops.ops)
+
+    def test_does_not_double_inject_when_already_present(self):
+        """On Postgres, autogenerate emits the expression index normally. The
+        hook must not duplicate it."""
+        from alembic.operations.ops import CreateIndexOp
+        from simple_module_db.migrations import make_process_revision_directives
+
+        meta = self._build_meta()
+        idx = next(iter(meta.tables["things"].indexes))
+        directives = self._build_directives("things", CreateIndexOp.from_index(idx))
+        make_process_revision_directives(meta)(None, None, directives)
+
+        index_ops = [op for op in directives[0].upgrade_ops.ops if isinstance(op, CreateIndexOp)]
+        assert len(index_ops) == 1
+
+    def test_ignores_column_based_indexes(self):
+        """Plain column indexes are already handled correctly by autogenerate;
+        the hook must not touch them."""
+        from alembic.operations.ops import CreateIndexOp
+        from simple_module_db.migrations import make_process_revision_directives
+
+        directives = self._build_directives("things")
+        make_process_revision_directives(self._build_meta(expression_index=False))(
+            None, None, directives
+        )
+
+        assert not any(isinstance(op, CreateIndexOp) for op in directives[0].upgrade_ops.ops)

host/migrations/env.py

@@ -11,7 +11,12 @@
 from logging.config import fileConfig
 
 from alembic import context
-from simple_module_db import build_module_metadata, make_include_object, render_item
+from simple_module_db import (
+    build_module_metadata,
+    make_include_object,
+    make_process_revision_directives,
+    render_item,
+)
 from simple_module_hosting.settings import Settings
 from sqlalchemy import engine_from_config, pool
 
@@ -31,6 +36,10 @@
 # host's user-added tables or framework internals.
 include_object = make_include_object(target_metadata)
 
+# Re-emit expression-based indexes (e.g. ``lower(email)``) that autogenerate
+# silently drops under SQLite — see make_process_revision_directives docstring.
+process_revision_directives = make_process_revision_directives(target_metadata)
+
 
 def _get_url() -> str:
     """Read database URL from settings, convert async to sync driver."""
@@ -48,6 +57,7 @@ def run_migrations_offline() -> None:
         dialect_opts={"paramstyle": "named"},
         include_object=include_object,
         render_item=render_item,
+        process_revision_directives=process_revision_directives,
     )
 
     with context.begin_transaction():
@@ -71,6 +81,7 @@ def run_migrations_online() -> None:
             target_metadata=target_metadata,
             include_object=include_object,
             render_item=render_item,
+            process_revision_directives=process_revision_directives,
         )
 
         with context.begin_transaction():

host/migrations/versions/41cf2c53660e_users_email_lower_index.py

@@ -0,0 +1,39 @@
+"""users_user lower(email) functional index
+
+Revision ID: 41cf2c53660e
+Revises: 3bf3f9db7f7f
+Create Date: 2026-05-12 00:00:00.000000
+
+The functional index ``ix_users_user_email_lower`` on ``lower(users_user.email)``
+backs the case-insensitive lookup used by ``UserDatabaseWithRoles.get_by_email``
+and ``users.bootstrap``. Autogenerate silently dropped it from the original
+initial-schema revision (``77162e7b184b``) because SQLAlchemy 2.0 can't reflect
+expression-based indexes under the SQLite dialect, so dev DBs are missing it.
+This revision back-fills the index unconditionally.
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "41cf2c53660e"
+down_revision: str | None = "3bf3f9db7f7f"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    # ``if_not_exists`` covers the case where a Postgres developer's original
+    # autogen run already emitted this index (only SQLite skips it).
+    op.create_index(
+        "ix_users_user_email_lower",
+        "users_user",
+        [sa.text("lower(email)")],
+        unique=False,
+        if_not_exists=True,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_users_user_email_lower", table_name="users_user", if_exists=True)