john/prompts
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

engine/session updates

Browse Source
This commit is contained in:
John Lancaster
2026-07-31 22:15:51 -05:00
parent cd11ea8255
commit 0dc06f72ca
12 changed files with 767 additions and 457 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/skills/async-fastapi-sqlmodel/references/testing.md
+65 -43
View File
@@ -1,94 +1,112 @@
# 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.
Use the same engine and session primitives in production and tests. Tests select a different URL and, when transaction isolation is required, bind a test session factory to one test-owned connection and outer 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 |
| Fast, serial application tests | `sqlite+aiosqlite://` | Per-test engine or connection-bound session factory over an 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
## Shared Construction Primitives
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.
Make the application factory accept a database URL or settings object. Production, workers, and ordinary integration tests enter the same [`database_scope()`](engine.md#one-engine-context-manager). Tests enter the lower-level `engine_scope()` only when they need direct engine or connection ownership for schema setup, an outer transaction, or engine-specific assertions:
```python
from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, create_async_engine
from sqlmodel.ext.asyncio.session import AsyncSession
from collections.abc import AsyncGenerator
import pytest_asyncio
from sqlalchemy.ext.asyncio import AsyncEngine
from .engine import engine_scope
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
@pytest_asyncio.fixture
async def test_engine(database_url: str) -> AsyncGenerator[AsyncEngine]:
async with engine_scope(database_url) as engine:
yield engine
```
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.
Production passes its `postgresql+asyncpg://...` URL to `database_scope()`. A local SQLite run passes `sqlite+aiosqlite:///./app.db`. Tests pass a dedicated test URL to `database_scope()` or `engine_scope()` and receive deterministic disposal when the context exits. Do not create an engine during module import: that makes it easy for tests to 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.
For tests that exercise code which commits, start an outer transaction on one test connection. Bind a test `SessionFactory` to that connection with `join_transaction_mode="create_savepoint"`. SQLAlchemy documents this as its test-suite pattern: sessions created by the factory resolve their commits through SAVEPOINTs while fixture teardown rolls back the outer transaction.
```python
from collections.abc import AsyncGeneratorr
from collections.abc import AsyncGenerator
import pytest_asyncio
from sqlalchemy.ext.asyncio import AsyncEngine
from sqlalchemy.ext.asyncio import async_sessionmaker
from sqlmodel.ext.asyncio.session import AsyncSession
from .session import SessionFactory
@pytest_asyncio.fixture
async def session(test_engine: AsyncEngine) -> AsyncGenerator[AsyncSession]:
async def session_factory(test_engine: AsyncEngine) -> AsyncGenerator[SessionFactory]:
async with test_engine.connect() as connection:
transaction = await connection.begin()
test_session = AsyncSession(
factory = async_sessionmaker(
bind=connection,
class_=AsyncSession,
expire_on_commit=False,
join_transaction_mode="create_savepoint",
)
try:
yield test_session
yield factory
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.
Each factory call still creates a distinct `AsyncSession`, matching [session factory mechanics](session.md#session-factory-mechanics). The factory belongs to the fixture's engine and outer transaction and must not escape either scope.
For service tests that need to pass a caller-owned active session into `transaction_scope(session=...)`, derive that session from the same factory:
```python
@pytest_asyncio.fixture
async def session(session_factory: SessionFactory) -> AsyncGenerator[AsyncSession]:
async with session_factory() as test_session:
await test_session.begin()
yield test_session
```
The explicit `begin()` satisfies the supplied-session contract from [session management](session.md#one-optional-ownership-helper). Session closure rolls back unfinished work; the outer connection transaction remains the final isolation boundary even if application code commits its SAVEPOINT.
For FastAPI request tests, override `get_session_factory`, not only `get_session`. Both `SessionDep` and `TransactionSessionDep` then retain their production ownership behavior while receiving the test-bound factory. Always remove the override after the test because [FastAPI dependency overrides](https://fastapi.tiangolo.com/advanced/testing-dependencies/) are stored in a mutable application-level dictionary.
```python
from collections.abc import Generator
import pytest
from fastapi import FastAPI
from sqlmodel.ext.asyncio.session import AsyncSession
from .fastapi import get_session_factory
from .session import SessionFactory
@pytest.fixture
def app_with_test_session(
app: FastAPI,
session: AsyncSession,
) -> FastAPI:
async def get_test_session() -> AsyncGenerator[AsyncSession]:
yield session
def app_with_test_database(app: FastAPI, session_factory: SessionFactory) -> Generator[FastAPI]:
def get_test_session_factory() -> SessionFactory:
return session_factory
app.dependency_overrides[get_session] = get_test_session
app.dependency_overrides[get_session_factory] = get_test_session_factory
try:
yield app
finally:
app.dependency_overrides.clear()
app.dependency_overrides.pop(get_session_factory, None)
```
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.
Construct `app` with test settings before lifespan starts so startup cannot resolve the production URL. The override changes request session creation; it does not prevent lifespan from entering its configured `database_scope()`.
The connection-bound factory is deliberately serial even though it creates distinct sessions: those sessions still share one connection and outer transaction. A test that verifies concurrently active sessions must use independent connections and a database target that supports them.
## SQLite Targets
@@ -99,19 +117,21 @@ Use `sqlite+aiosqlite://` for a fresh in-memory database when the test runs all
Create the schema and dispose the engine deterministically:
```python
from collections.abc import AsyncGenerator
import pytest_asyncio
from sqlalchemy.ext.asyncio import AsyncEngine
from sqlmodel import SQLModel
from .engine import engine_scope
@pytest_asyncio.fixture
async def test_engine() -> AsyncGenerator[AsyncEngine]:
engine, _ = create_database("sqlite+aiosqlite://")
async with engine.begin() as connection:
await connection.run_sync(SQLModel.metadata.create_all)
try:
async with engine_scope("sqlite+aiosqlite://") as engine:
async with engine.begin() as connection:
await connection.run_sync(SQLModel.metadata.create_all)
yield engine
finally:
await engine.dispose()
```
### Concurrent in-memory tests
@@ -140,8 +160,9 @@ For both SQLite forms, enable and test the constraints your application depends
## 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.
- Production PostgreSQL, local SQLite, and in-memory SQLite all use `database_scope()` unless a test explicitly needs lower-level engine or connection ownership.
- Every test owns its override, session factory, connection, transaction, session, and engine cleanup.
- Request tests override `get_session_factory`, preserving both read-session and transactional-session dependency behavior.
- 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.
@@ -151,4 +172,5 @@ For both SQLite forms, enable and test the constraints your application depends
- [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/)
- [SQLModel testing with FastAPI](https://sqlmodel.tiangolo.com/tutorial/fastapi/tests/)
- [pytest-asyncio fixtures](https://pytest-asyncio.readthedocs.io/en/stable/how-to-guides/index.html)