diff --git a/docs/skills/async-fastapi-sqlmodel/SKILL.md b/docs/skills/async-fastapi-sqlmodel/SKILL.md index be17243..2bcfbe7 100644 --- a/docs/skills/async-fastapi-sqlmodel/SKILL.md +++ b/docs/skills/async-fastapi-sqlmodel/SKILL.md @@ -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 diff --git a/docs/skills/async-fastapi-sqlmodel/references/crud.md b/docs/skills/async-fastapi-sqlmodel/references/crud.md index 6b90c54..6dbf4ab 100644 --- a/docs/skills/async-fastapi-sqlmodel/references/crud.md +++ b/docs/skills/async-fastapi-sqlmodel/references/crud.md @@ -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: diff --git a/docs/skills/async-fastapi-sqlmodel/references/index.md b/docs/skills/async-fastapi-sqlmodel/references/index.md index 1066fbf..476b7e9 100644 --- a/docs/skills/async-fastapi-sqlmodel/references/index.md +++ b/docs/skills/async-fastapi-sqlmodel/references/index.md @@ -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 | --- diff --git a/docs/skills/async-fastapi-sqlmodel/references/session.md b/docs/skills/async-fastapi-sqlmodel/references/session.md index 56c420e..3f5b799 100644 --- a/docs/skills/async-fastapi-sqlmodel/references/session.md +++ b/docs/skills/async-fastapi-sqlmodel/references/session.md @@ -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 diff --git a/docs/skills/async-fastapi-sqlmodel/references/sqlmodel.md b/docs/skills/async-fastapi-sqlmodel/references/sqlmodel.md index 7765dbc..d307348 100644 --- a/docs/skills/async-fastapi-sqlmodel/references/sqlmodel.md +++ b/docs/skills/async-fastapi-sqlmodel/references/sqlmodel.md @@ -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. --- diff --git a/docs/skills/async-fastapi-sqlmodel/references/testing.md b/docs/skills/async-fastapi-sqlmodel/references/testing.md new file mode 100644 index 0000000..0f3b6f8 --- /dev/null +++ b/docs/skills/async-fastapi-sqlmodel/references/testing.md @@ -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/) \ No newline at end of file