fastapi updates

2026-07-30 01:28:39 -05:00
parent 3abafc4850
commit b6f109cf91
6 changed files with 183 additions and 11 deletions
@@ -110,6 +110,14 @@ Pool sizing, overflow, recycle, pre-ping, isolation, statement timeouts, and hea

 See [observability and resilience](references/observability.md).

+### Test through the production seam
+
+Keep the production engine and session-factory construction path intact in tests. Select a dedicated PostgreSQL, local SQLite, or in-memory SQLite URL at that seam, then override the request-session dependency only for the test lifetime. Use a test-scoped outer transaction with SAVEPOINT-backed session commits when application code calls `commit()`; it exercises normal transaction behavior while cleanup remains deterministic.
+
+In-memory SQLite is suitable for serial tests. For multiple simultaneous sessions, use a named shared-cache SQLite URL or a temporary file, and retain PostgreSQL integration coverage for PostgreSQL-specific behavior.
+
+See [database testing and fixture data](references/testing.md).
+
 ## Reference Map

 | Concept | Reference |
@@ -123,6 +131,7 @@ See [observability and resilience](references/observability.md).
 | Observability and resilience | [Observability reference](references/observability.md) |
 | SQLModel-first modeling | [SQLModel integration reference](references/sqlmodel.md) |
 | CRUD repository and standalone functions | [Basic CRUD reference](references/crud.md) |
+| Test database selection and fixture data | [Database testing reference](references/testing.md) |

 ## Canonical Composition Pattern

@@ -133,7 +142,8 @@ from contextlib import AsyncExitStack, asynccontextmanager
 from collections.abc import AsyncIterator

 from fastapi import FastAPI
-from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+from sqlmodel.ext.asyncio.session import AsyncSession

@asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncIterator[None]:
@@ -141,7 +151,11 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
        engine = create_async_engine(settings.database_url)
        stack.push_async_callback(engine.dispose)

-        session_factory = async_sessionmaker(engine, expire_on_commit=False)
+        session_factory = async_sessionmaker(
+          engine,
+          class_=AsyncSession,
+          expire_on_commit=False,
+        )
        app.state.session_factory = session_factory
        yield

@@ -173,6 +187,7 @@ When reviewing code, verify:
 - Relationship and deferred-column access cannot surprise the event loop with implicit I/O.
 - Pool and timeout settings are justified by deployment behavior.
 - Tests exercise rollback, cleanup, concurrency, and lifespan behavior where relevant.
+- Tests use a dedicated database target and preserve production session mechanics.

 ## Anti-Patterns to Flag

@@ -61,8 +61,8 @@ This reference uses direct field arguments and full-update semantics to keep the
 Functions are the simplest default when grouping state or behavior in an object adds no value. Each function is a complete operation boundary: it can run standalone by resolving the cached factory from `database_url`, or compose into a caller-owned scope through `session`.

 ```python
-from sqlalchemy.ext.asyncio import AsyncSession
 from sqlmodel import select
+from sqlmodel.ext.asyncio.session import AsyncSession

 from .session import session_scope
 from .session import transaction_scope
@@ -170,7 +170,7 @@ Update and delete load the row through the same session that mutates it. This av
 A repository can provide a stable domain-facing interface when several callers need the same grouped operations. It stores repeatable database configuration, never a mutable session. Every method delegates to the analogous function and exposes the same optional-session contract.

 ```python
-from sqlalchemy.ext.asyncio import AsyncSession
+from sqlmodel.ext.asyncio.session import AsyncSession

 class WidgetRepository:
    def __init__(self, database_url: str) -> None:
@@ -15,6 +15,7 @@ Purpose: concept registry for the principles, mechanics, and implementation guid
 | Observability and resilience | [observability.md](observability.md) | adopted | mandatory | platform/backend | 2026-06-17 |
 | SQLModel modeling and async boundaries | [sqlmodel.md](sqlmodel.md) | adopted | mandatory | platform/backend | 2026-07-26 |
 | Basic CRUD repository and functions | [crud.md](crud.md) | adopted | advisory | platform/backend | 2026-07-26 |
+| Test database targets and fixture data | [testing.md](testing.md) | adopted | mandatory | platform/backend | 2026-07-30 |

 ---

@@ -81,7 +81,8 @@ Cache it by the application-owned engine so repeated composition calls return th
 ```python
 from functools import cache

-from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+from sqlalchemy.ext.asyncio import async_sessionmaker
+from sqlmodel.ext.asyncio.session import AsyncSession

 from .engine import dispose_engine
 from .engine import get_engine
@@ -226,7 +227,7 @@ Pass the database URL to repository constructors. The repository stores repeatab

 ```python
 from sqlalchemy import select
-from sqlalchemy.ext.asyncio import AsyncSession
+from sqlmodel.ext.asyncio.session import AsyncSession


 async def find_item(session: AsyncSession, item_id: int) -> Item | None:
@@ -280,8 +281,8 @@ from collections.abc import AsyncIterator

 from fastapi import Depends
 from fastapi import Request
-from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.ext.asyncio import async_sessionmaker
+from sqlmodel.ext.asyncio.session import AsyncSession


 type SessionFactory = async_sessionmaker[AsyncSession]
@@ -302,7 +303,7 @@ Route usage:

 ```python
 from fastapi import APIRouter, Depends
-from sqlalchemy.ext.asyncio import AsyncSession
+from sqlmodel.ext.asyncio.session import AsyncSession

 from .session import get_db_session

@@ -34,7 +34,7 @@ SQLModel is designed for FastAPI, built on Pydantic and SQLAlchemy, and intended
 ## Rules

 - Default to SQLModel for new table models and API data models.
- Keep SQLAlchemy async primitives as the runtime base: `create_async_engine`, `async_sessionmaker`, and `AsyncSession`.
+- Keep SQLAlchemy engine and factory primitives as the runtime base: `create_async_engine` and `async_sessionmaker`. For SQLModel applications, use SQLModel's `AsyncSession` wrapper so its typed `exec()` API remains available.
 - Keep transaction and session ownership policies identical whether models are SQLAlchemy Declarative or SQLModel.
 - Use SQLModel inheritance to share validated fields while keeping table, create, update, and public contracts distinct where their semantics differ.
 - Use SQLAlchemy declarative models only for a concrete unsupported mapping or third-party constraint; document the reason.
@@ -72,8 +72,9 @@ class UserRead(UserBase):
 ### Pattern B: Keep SQLModel models with the async runtime

 ```python
-from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
 from sqlmodel import select
+from sqlmodel.ext.asyncio.session import AsyncSession

 engine = create_async_engine(settings.database_url, pool_pre_ping=True)
 session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
@@ -82,7 +83,7 @@ async with session_factory() as session:
    users = (await session.scalars(select(User))).all()
 ```

-`sqlmodel.select()` keeps SQLModel's typing-oriented statement construction, while `AsyncSession.scalars()` and the surrounding lifecycle come from SQLAlchemy.
+`sqlmodel.select()` keeps SQLModel's typing-oriented statement construction, and SQLModel's `AsyncSession` adds typed `exec()` results while retaining SQLAlchemy's async lifecycle and transaction behavior. Import `AsyncSession` from `sqlmodel.ext.asyncio.session` when working with SQLModel models; use SQLAlchemy's `AsyncSession` only when the code intentionally has no SQLModel dependency.

 ---

@@ -0,0 +1,154 @@
+# Testing Database Targets and Data
+
+Use the same application database construction path in production and tests. Tests select a different URL and bind their request-session dependency to a test-scoped transaction; they do not replace repositories, services, or SQLAlchemy mechanics with mocks.
+
+## Decision Table
+
+| Test need | Database target | Isolation approach | What it proves |
+|---|---|---|---|
+| Fast, serial application tests | `sqlite+aiosqlite://` | Per-test engine or outer transaction | ORM mappings and ordinary application behavior |
+| Async code using multiple simultaneous sessions | Named SQLite shared-cache URL or temporary SQLite file | Per-test schema or cleanup strategy | Concurrent-session behavior without a database server |
+| PostgreSQL-specific behavior | Dedicated PostgreSQL test database | Per-test outer transaction and SAVEPOINT | SQL, constraints, types, locking, and migrations that SQLite cannot represent |
+
+SQLite is a useful fast target, not a drop-in PostgreSQL substitute. Keep a small PostgreSQL integration suite for PostgreSQL-specific queries, extensions, row locking, JSON semantics, collations, isolation, and migration validation.
+
+## One Construction Path
+
+Make the application factory accept a database URL or settings object, and keep engine and session-factory construction in one function. The only test-specific inputs should be the URL and, for request tests, the session dependency override.
+
+```python
+from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, create_async_engine
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+
+def create_database(
+    database_url: str,
+) -> tuple[AsyncEngine, async_sessionmaker[AsyncSession]]:
+    engine = create_async_engine(database_url)
+    session_factory = async_sessionmaker(
+        engine,
+        class_=AsyncSession,
+        expire_on_commit=False,
+    )
+    return engine, session_factory
+```
+
+Production passes its `postgresql+asyncpg://...` URL to `create_database()`. A local SQLite run passes `sqlite+aiosqlite:///./app.db`. Tests pass a dedicated test URL to the same function. Do not create an engine during module import: that makes it easy for tests to accidentally retain the production URL before an override is applied.
+
+Use migrations to provision an integration database when migrations are part of the release contract. `metadata.create_all()` is appropriate for focused ORM tests only when it accurately represents the schema under test. Import all table models before creating metadata; [SQLModel documents that model-registration order matters](https://sqlmodel.tiangolo.com/tutorial/fastapi/tests/#import-table-models).
+
+## Transactional Async Fixture
+
+For tests that exercise code which calls `commit()`, start an outer transaction on one test connection. Bind the test `AsyncSession` to that connection and use `join_transaction_mode="create_savepoint"`. SQLAlchemy documents this as its test-suite pattern: session commits resolve a SAVEPOINT while fixture teardown rolls back the outer transaction.
+
+```python
+from collections.abc import AsyncIterator
+
+import pytest_asyncio
+from sqlalchemy.ext.asyncio import AsyncEngine
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+
+@pytest_asyncio.fixture
+async def session(test_engine: AsyncEngine) -> AsyncIterator[AsyncSession]:
+    async with test_engine.connect() as connection:
+        transaction = await connection.begin()
+        test_session = AsyncSession(
+            bind=connection,
+            expire_on_commit=False,
+            join_transaction_mode="create_savepoint",
+        )
+        try:
+            yield test_session
+        finally:
+            await test_session.close()
+            await transaction.rollback()
+```
+
+Use the test session through the normal FastAPI dependency seam, and always remove the override after the test. [FastAPI dependency overrides](https://fastapi.tiangolo.com/advanced/testing-dependencies/) are an application-level dictionary, so leaving one installed leaks test state.
+
+```python
+import pytest
+from fastapi import FastAPI
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+
+@pytest.fixture
+def app_with_test_session(
+    app: FastAPI,
+    session: AsyncSession,
+) -> FastAPI:
+    async def get_test_session() -> AsyncIterator[AsyncSession]:
+        yield session
+
+    app.dependency_overrides[get_session] = get_test_session
+    try:
+        yield app
+    finally:
+        app.dependency_overrides.clear()
+```
+
+This fixture is deliberately serial: one mutable `AsyncSession` must not serve concurrent tasks. A test that verifies concurrently active sessions should create independent sessions from a factory and use a database target that supports independent connections.
+
+## SQLite Targets
+
+### Serial in-memory tests
+
+Use `sqlite+aiosqlite://` for a fresh in-memory database when the test runs all database work serially. SQLAlchemy's `aiosqlite` dialect uses a single-connection `StaticPool` for this target, so all sessions share one SQLite transaction state. One session's rollback can discard another session's uncommitted work.
+
+Create the schema and dispose the engine deterministically:
+
+```python
+import pytest_asyncio
+from sqlalchemy.ext.asyncio import AsyncEngine
+
+
+@pytest_asyncio.fixture
+async def test_engine() -> AsyncIterator[AsyncEngine]:
+    engine, _ = create_database("sqlite+aiosqlite://")
+    async with engine.begin() as connection:
+        await connection.run_sync(SQLModel.metadata.create_all)
+    try:
+        yield engine
+    finally:
+        await engine.dispose()
+```
+
+### Concurrent in-memory tests
+
+Do not use the default `:memory:` target for tests that have multiple active sessions or tasks. Use a named shared-cache database instead, with a name unique to the test process:
+
+```text
+sqlite+aiosqlite:///file:test-suite?mode=memory&cache=shared&uri=true
+```
+
+This lets connections share the same in-memory database while retaining independent transaction state. A temporary file URL such as `sqlite+aiosqlite:////tmp/test.db` is often simpler when test isolation or cleanup tooling already manages files.
+
+For both SQLite forms, enable and test the constraints your application depends on. SQLite foreign-key enforcement is disabled by default, and its transaction behavior has driver-specific differences. Keep PostgreSQL integration coverage for behavior that SQLite cannot faithfully model.
+
+## Test Data Practices
+
+- Build only the data a test needs, through named factory functions or pytest fixtures rather than a large global seed.
+- Give each fixture a domain meaning, such as `active_account`, `expired_subscription`, or `admin_user`; avoid opaque rows with unexplained defaults.
+- Set values relevant to the assertion explicitly, including timestamps, permissions, statuses, and unique identifiers. Use fixed clocks or injected clock values instead of the wall clock.
+- Construct object graphs through relationships, then `await session.flush()` before reading generated identifiers or passing foreign keys onward. `flush()` exercises database constraints without ending the test transaction.
+- Seed prerequisite data before creating a client request. Let the endpoint own the mutation being asserted; do not pre-insert the row that the endpoint is supposed to create.
+- Use `commit()` in fixture setup only when the test specifically needs to prove post-commit behavior. With the transactional fixture, this remains isolated through the outer rollback.
+- Keep shared reference data immutable and explicit. If it must be reused for performance, load it once into a dedicated test database and reset all mutable tables between tests; never depend on test order.
+- Include both valid and constraint-breaking graphs where a behavior depends on foreign keys, uniqueness, nullability, or cascading deletes. SQLite-only tests should not be the sole evidence for PostgreSQL constraints.
+
+## Completion Checks
+
+- A test run cannot reach the production URL; production credentials are absent from the test environment.
+- Production PostgreSQL, local SQLite, and in-memory SQLite all use the same engine/session-factory construction path.
+- Every test owns its override, connection, transaction, session, and engine cleanup.
+- Test data is deterministic, minimal, and expresses the scenario under test.
+- PostgreSQL integration tests cover every PostgreSQL-specific contract and run against migrations where migrations are shipped.
+
+## Sources
+
+- [SQLAlchemy: joining a session into an external transaction](https://docs.sqlalchemy.org/en/21/orm/session_transaction.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites)
+- [SQLAlchemy asyncio extension](https://docs.sqlalchemy.org/en/21/orm/extensions/asyncio.html)
+- [SQLAlchemy SQLite dialect and async in-memory pooling](https://docs.sqlalchemy.org/en/21/dialects/sqlite.html#using-a-memory-database-with-multiple-coroutines)
+- [FastAPI dependency overrides](https://fastapi.tiangolo.com/advanced/testing-dependencies/)
+- [SQLModel testing with FastAPI](https://sqlmodel.tiangolo.com/tutorial/fastapi/tests/)