perf(testing): copy a per-process SQLite schema into test databases; cache the Alembic revision map

CHANGELOG.md

@@ -14,6 +14,11 @@ All notable changes to this project are documented in this file. The format is b
 
 ## [Unreleased]
 
+### Changed
+
+- Faster test fixtures: `simple_module_test` builds the in-memory SQLite schema once per process and copies it into each test's database with SQLite's backup API, instead of running `create_all` for every `app` and `db_session` fixture. A database that already has tables is left alone. `make test-py` drops from 324 s to 256 s on 8 workers, and fixture setup time falls by about a third.
+- `simple_module_hosting.migrations.script_directory` reuses the loaded Alembic revision map while the revision files are unchanged (any added, removed or edited file reloads it). Each app boot previously loaded every migration file twice. The background-tasks Celery app also caches its entry-point scan for installed task packages.
+
 ## [0.0.44] — 2026-11-06
 
 ### Fixed

docs/testing/fixtures.md

@@ -15,7 +15,7 @@ The `simple_module_test` plugin provides app-level fixtures that every test dire
 | `authenticated_client` | function | `httpx.AsyncClient` | Admin-scoped HTTP calls — carries a signed session cookie. |
 | `captured_mail` | function | `CapturedMail` | Asserting on outgoing mail: installs a recording sender (`.messages`, `.to(address)`) as the process mail sender. |
 
-All fixtures are **function-scoped** — each test gets its own. The SQLite database is in-memory; tearing down and re-creating is cheap.
+All fixtures are **function-scoped** — each test gets its own. The SQLite database is in-memory; tearing down and re-creating is cheap. The first schema each process builds is kept as a template and copied into every later empty in-memory database with SQLite's backup API, so `CREATE TABLE` runs once per worker rather than once per test (`simple_module_test._sqlite_template`). A database that already has tables is never overwritten.
 
 ## Postgres
 
@@ -43,7 +43,7 @@ def test_settings_defaults(settings):
 
 ## `db_session`
 
-The workhorse. Creates a fresh in-memory DB, runs `CREATE TABLE` for every module's tables (iterating `all_module_bases` and calling `base.metadata.create_all`), stamps `alembic_version` at head, and yields an `AsyncSession`.
+The workhorse. Creates a fresh in-memory DB, gives it every module's tables (iterating `all_module_bases` and calling `base.metadata.create_all` the first time in a process, copying that schema afterwards), stamps `alembic_version` at head, and yields an `AsyncSession`.
 
 ```python
 from decimal import Decimal

framework/hosting/simple_module_hosting/migrations.py

@@ -5,6 +5,10 @@
 import logging
 import os
 from pathlib import Path
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from alembic.script import ScriptDirectory
 
 logger = logging.getLogger(__name__)
 
@@ -29,17 +33,48 @@ def default_alembic_ini() -> str:
     return str(Path(root) / _ALEMBIC_INI_RELATIVE) if root else _ALEMBIC_INI_RELATIVE
 
 
-def script_directory(alembic_ini_path: str | None = None):
+# ini path -> (revision files and their mtimes, ScriptDirectory with its revision map loaded).
+_SCRIPTS: dict[str, tuple[tuple, ScriptDirectory]] = {}
+
+
+def _revision_files(script: ScriptDirectory) -> tuple:
+    # ``_version_locations`` is alembic's resolved list of version directories (there
+    # is no public accessor for more than one location).
+    files = []
+    for location in script._version_locations:
+        for path in Path(location).rglob("*.py"):
+            try:
+                stat = path.stat()
+            except FileNotFoundError:  # removed while scanning: the next call sees the new set
+                continue
+            files.append((str(path), stat.st_mtime_ns, stat.st_size))
+    return tuple(sorted(files))
+
+
+def script_directory(alembic_ini_path: str | None = None) -> ScriptDirectory:
     """Build the ``ScriptDirectory`` for ``alembic_ini_path``.
 
     Shared by every caller that needs to walk or query alembic revisions
     (boot-time head check, the in-app Doctor screen) so the config/ini-path
     construction has exactly one source of truth.
+
+    Reused while the revision files are unchanged: loading the revision map
+    imports every migration file, which every app build paid for twice. A file
+    added, removed or edited (``smpy migrations prepare``, a new revision in
+    development) builds a fresh one.
     """
     from alembic.config import Config as AlembicConfig
     from alembic.script import ScriptDirectory
 
-    return ScriptDirectory.from_config(AlembicConfig(alembic_ini_path or default_alembic_ini()))
+    ini_path = alembic_ini_path or default_alembic_ini()
+    script = ScriptDirectory.from_config(AlembicConfig(ini_path))
+    files = _revision_files(script)
+    key = str(Path(ini_path).resolve())
+    cached = _SCRIPTS.get(key)
+    if cached is not None and cached[0] == files:
+        return cached[1]
+    _SCRIPTS[key] = (files, script)
+    return script
 
 
 def resolve_head_revisions(alembic_ini_path: str | None = None) -> tuple[str, ...]:

framework/hosting/tests/test_migration_script_cache.py

@@ -0,0 +1,79 @@
+"""``script_directory`` reuses the loaded revision map until a revision file changes."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import pytest
+from simple_module_hosting import migrations
+from simple_module_hosting.migrations import resolve_head_revisions, script_directory
+
+_INI = """\
+[alembic]
+script_location = %(here)s/migrations
+"""
+
+_REVISION = """\
+revision = "{rev}"
+down_revision = {down}
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass
+"""
+
+
+@pytest.fixture
+def project(tmp_path: Path, monkeypatch) -> Path:
+    monkeypatch.setattr(migrations, "_SCRIPTS", {})
+    (tmp_path / "migrations" / "versions").mkdir(parents=True)
+    (tmp_path / "alembic.ini").write_text(_INI)
+    _write(tmp_path, "a1", None)
+    return tmp_path
+
+
+def _write(project: Path, rev: str, down: str | None) -> Path:
+    path = project / "migrations" / "versions" / f"{rev}_rev.py"
+    path.write_text(_REVISION.format(rev=rev, down=repr(down)))
+    return path
+
+
+def test_an_unchanged_history_reuses_the_loaded_script(project: Path) -> None:
+    ini = str(project / "alembic.ini")
+    assert script_directory(ini) is script_directory(ini)
+
+
+def test_a_new_revision_is_seen(project: Path) -> None:
+    ini = str(project / "alembic.ini")
+    assert resolve_head_revisions(ini) == ("a1",)
+
+    _write(project, "b2", "a1")
+
+    assert resolve_head_revisions(ini) == ("b2",)
+
+
+def test_an_edited_revision_is_seen(project: Path) -> None:
+    ini = str(project / "alembic.ini")
+    first = script_directory(ini)
+    path = _write(project, "a1", None)
+    stat = path.stat()
+    os.utime(path, ns=(stat.st_atime_ns, stat.st_mtime_ns + 1_000_000))
+
+    assert script_directory(ini) is not first
+
+
+def test_a_removed_revision_is_seen(project: Path) -> None:
+    ini = str(project / "alembic.ini")
+    second = _write(project, "b2", "a1")
+    assert resolve_head_revisions(ini) == ("b2",)
+
+    second.unlink()
+
+    assert resolve_head_revisions(ini) == ("a1",)

framework/testing/simple_module_test/_schema.py

@@ -82,13 +82,15 @@ def _sync_create_all(sync_conn):
 
             await conn.run_sync(_sync_create_all)
             # Part of the build, so a reused Postgres schema's fingerprint covers it.
-            if heads:
-                await conn.execute(
-                    text(
-                        "CREATE TABLE IF NOT EXISTS alembic_version "
-                        "(version_num VARCHAR(32) NOT NULL PRIMARY KEY)"
-                    )
+            # Created even with no heads to stamp: the build is copied into later
+            # SQLite databases (``_sqlite_template``), and one that resolved no heads
+            # must not hand a schema without the table to a test that stamps it.
+            await conn.execute(
+                text(
+                    "CREATE TABLE IF NOT EXISTS alembic_version "
+                    "(version_num VARCHAR(32) NOT NULL PRIMARY KEY)"
                 )
+            )
 
     tables = [table for base in bases for table in base.metadata.tables.values()]
     # Postgres reuses the schema built for an earlier test, emptied of rows.

framework/testing/simple_module_test/_sqlite_template.py

@@ -0,0 +1,74 @@
+"""Copy a per-process template into each test's in-memory SQLite database.
+
+Every ``app`` and ``db_session`` fixture gets a fresh ``:memory:`` database and
+used to run ``create_all`` on it: DDL for every module's tables and indexes,
+about a third of the fixture's cost. The schema is the same for every test in
+a process, so the first build is copied into a plain ``sqlite3`` connection and
+later databases are filled from it with SQLite's backup API, which copies the
+pages instead of compiling and running the DDL again.
+
+Each test still gets its own database: the copy only replaces an *empty* one.
+A database that already has tables (a test calling the schema helper twice)
+gets the ordinary ``create_all``, which leaves its rows alone. The template is
+rebuilt whenever the set of tables changes, such as a test declaring a model.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+from typing import Any
+
+from sqlalchemy import text
+
+_template: tuple[frozenset, sqlite3.Connection] | None = None
+
+
+def is_memory_sqlite(engine: Any) -> bool:
+    """Whether a SQLite *engine* is an in-memory database (the caller checked the dialect)."""
+    return engine.url.database in (None, "", ":memory:")
+
+
+def _key(tables: list[Any]) -> frozenset:
+    return frozenset(
+        (t.fullname, tuple(t.columns.keys()), tuple(sorted(i.name or "" for i in t.indexes)))
+        for t in tables
+    )
+
+
+async def _driver_connection(conn: Any) -> Any:
+    """The ``aiosqlite.Connection`` under an SQLAlchemy async connection."""
+    raw = await conn.get_raw_connection()
+    return raw.driver_connection
+
+
+async def build_from_template(engine: Any, tables: list[Any], build: Any) -> None:
+    """Give *engine* the schema ``build`` creates, copying it when possible."""
+    global _template
+    async with engine.connect() as conn:
+        has_tables = await conn.scalar(text("SELECT count(*) FROM sqlite_master"))
+    if has_tables:
+        await build(engine)
+        return
+
+    key = _key(tables)
+    if _template is not None and _template[0] == key:
+        source = _template[1]
+        async with engine.connect() as conn:
+            target = await _driver_connection(conn)
+            # No public path: aiosqlite's ``backup`` only copies *out* of its own
+            # connection, and a template held as an aiosqlite connection would add a
+            # non-daemon worker thread that must be closed at exit. So run the stdlib
+            # ``backup`` on the target's worker thread, which owns ``_conn``, via
+            # the private ``_execute``/``_conn`` (present across the ``aiosqlite>=0.20,<0.23``
+            # range ``simple_module_db`` allows; the tests here fail loudly if they go).
+            await target._execute(source.backup, target._conn)
+        return
+
+    await build(engine)
+    # Read from aiosqlite worker threads of many connections, never concurrently.
+    copy = sqlite3.connect(":memory:", check_same_thread=False)
+    async with engine.connect() as conn:
+        await (await _driver_connection(conn)).backup(copy)
+    if _template is not None:
+        _template[1].close()
+    _template = (key, copy)

framework/testing/simple_module_test/database.py

@@ -23,6 +23,8 @@
 from sqlalchemy.engine import make_url
 from sqlalchemy.pool import NullPool
 
+from simple_module_test._sqlite_template import build_from_template, is_memory_sqlite
+
 TEST_DATABASE_ENV = "SM_TEST_DATABASE_URL"
 SQLITE_MEMORY = "sqlite+aiosqlite:///:memory:"
 XDIST_WORKER_ENV = "PYTEST_XDIST_WORKER"
@@ -146,15 +148,20 @@ async def reset_schema_once(engine: Any) -> None:
 async def prepare_schema_once(engine: Any, tables: list[Any], build: Any) -> None:
     """Give this test empty ``tables``, at most once per test (see ``begin_test``).
 
-    SQLite: ``build(engine)``, since every engine is a fresh in-memory database.
+    SQLite: every engine is a fresh database; an in-memory one is filled from a
+    per-process copy of the first build (``_sqlite_template``), a file one runs
+    ``build(engine)``.
     Postgres: when this process already built the schema and the catalog still
     matches it, delete every row and restart the sequences (~0.03 s) instead of
     dropping and rebuilding it (~0.25 s). Anything else (the first test, a
     direct ``reset_schema`` call, a test's own DDL) resets and rebuilds.
     """
     global _reset_for, _built_schema
     if engine.dialect.name == "sqlite":
-        await build(engine)
+        if is_memory_sqlite(engine):
+            await build_from_template(engine, tables, build)
+        else:
+            await build(engine)
         return
     if _current_test is not None and _reset_for is _current_test:
         return