swapped docs symlink
This commit is contained in:
John Lancaster
@@ -0,0 +1,294 @@
|
||||
# Basic CRUD Repository and Functions
|
||||
|
||||
!!! info "Primary sources"
|
||||
- [SQLModel create-data tutorial](https://sqlmodel.tiangolo.com/tutorial/fastapi/multiple-models/)
|
||||
- [SQLModel update-data tutorial](https://sqlmodel.tiangolo.com/tutorial/fastapi/update-extra-data/)
|
||||
- [SQLModel select tutorial](https://sqlmodel.tiangolo.com/tutorial/select/)
|
||||
- [SQLAlchemy `AsyncSession` API](https://docs.sqlalchemy.org/en/21/orm/extensions/asyncio.html#sqlalchemy.ext.asyncio.AsyncSession)
|
||||
- [`nicegui-db` service functions](https://forgejo.john-stream.com/john/nicegui-db/src/commit/126bc26ad8635a86bacf684d7bda409230347597/src/nicegui_db/services/my_table.py)
|
||||
|
||||
??? abstract "Decision metadata"
|
||||
- Status: adopted
|
||||
- Decision level: advisory
|
||||
- Applies to: api-runtime, workers, tests
|
||||
- Last reviewed: 2026-08-06
|
||||
|
||||
---
|
||||
|
||||
## Purpose
|
||||
|
||||
Show a small SQLModel CRUD layer in two forms:
|
||||
|
||||
- independent functions for convenient standalone or composed operations;
|
||||
- a repository object that groups those functions behind one domain-oriented interface.
|
||||
|
||||
Template-style public functions use `@with_session` and accept an optional `AsyncSession`. When the argument is omitted, the decorator resolves the cached session factory and owns a short-lived session. When supplied, the function borrows the session without controlling its lifetime or transaction. The decorator does not commit, so standalone writes need a visible transaction strategy; repository methods remain explicit-session operations for predictable composition.
|
||||
|
||||
Use the same vocabulary at every layer:
|
||||
|
||||
| Operation | Function | Repository method | Scope when session is omitted | Missing-row result |
|
||||
|---|---|---|---|---|
|
||||
| Create | `create_widget()` | `create()` | Owned session; no implicit commit | Not applicable |
|
||||
| Read one | `get_widget()` | `get()` | Owned session | `None` |
|
||||
| Read many | `list_widgets()` | `list()` | Owned session | Empty list |
|
||||
| Update | `update_widget()` | `update()` | Owned session; no implicit commit | `None` |
|
||||
| Delete | `delete_widget()` | `delete()` | Owned session; no implicit commit | `None` |
|
||||
|
||||
Functions and repository methods both put domain arguments first. Database configuration and sessions are keyword-only infrastructure arguments. This keeps call sites analogous and makes ownership choices visible.
|
||||
|
||||
---
|
||||
|
||||
## Models
|
||||
|
||||
Start with one table model when the application does not need distinct persistence and API schemas.
|
||||
|
||||
```python
|
||||
from sqlmodel import Field
|
||||
from sqlmodel import SQLModel
|
||||
|
||||
|
||||
class Widget(SQLModel, table=True):
|
||||
id: int | None = Field(default=None, primary_key=True)
|
||||
name: str = Field(index=True)
|
||||
description: str | None = None
|
||||
```
|
||||
|
||||
This reference uses direct field arguments and full-update semantics to keep the CRUD mechanics visible. Introduce separate create, update, or public schemas only when an API boundary needs different validation, field visibility, or partial-update behavior. See [SQLModel integration](sqlmodel.md) for that larger modeling pattern.
|
||||
|
||||
---
|
||||
|
||||
## Independent CRUD Functions
|
||||
|
||||
Functions are the simplest default when grouping state or behavior in an object adds no value. Decorate public service functions when both standalone reads and explicit composition are useful. The assertion narrows the optional type after decorator injection and catches accidental explicit `None` calls.
|
||||
|
||||
```python
|
||||
from sqlmodel import select
|
||||
from sqlmodel.ext.asyncio.session import AsyncSession
|
||||
|
||||
from .session import with_session
|
||||
|
||||
|
||||
@with_session
|
||||
async def create_widget(
|
||||
name: str,
|
||||
description: str | None = None,
|
||||
*,
|
||||
session: AsyncSession | None = None,
|
||||
) -> Widget:
|
||||
assert session is not None, "Session must be provided by with_session decorator"
|
||||
widget = Widget(name=name, description=description)
|
||||
session.add(widget)
|
||||
await session.flush()
|
||||
return widget
|
||||
|
||||
|
||||
@with_session
|
||||
async def get_widget(
|
||||
widget_id: int,
|
||||
*,
|
||||
session: AsyncSession | None = None,
|
||||
) -> Widget | None:
|
||||
assert session is not None, "Session must be provided by with_session decorator"
|
||||
return await session.get(Widget, widget_id)
|
||||
|
||||
|
||||
@with_session
|
||||
async def list_widgets(
|
||||
*,
|
||||
offset: int = 0,
|
||||
limit: int = 100,
|
||||
session: AsyncSession | None = None,
|
||||
) -> list[Widget]:
|
||||
assert session is not None, "Session must be provided by with_session decorator"
|
||||
if offset < 0:
|
||||
raise ValueError("offset must be non-negative")
|
||||
if not 1 <= limit <= 100:
|
||||
raise ValueError("limit must be between 1 and 100")
|
||||
|
||||
statement = select(Widget).order_by(Widget.id).offset(offset).limit(limit)
|
||||
return list(await session.scalars(statement))
|
||||
|
||||
|
||||
@with_session
|
||||
async def update_widget(
|
||||
widget_id: int,
|
||||
name: str,
|
||||
description: str | None,
|
||||
*,
|
||||
session: AsyncSession | None = None,
|
||||
) -> Widget | None:
|
||||
assert session is not None, "Session must be provided by with_session decorator"
|
||||
widget = await session.get(Widget, widget_id)
|
||||
if widget is None:
|
||||
return None
|
||||
|
||||
widget.name = name
|
||||
widget.description = description
|
||||
await session.flush()
|
||||
return widget
|
||||
|
||||
|
||||
@with_session
|
||||
async def delete_widget(
|
||||
widget_id: int,
|
||||
*,
|
||||
session: AsyncSession | None = None,
|
||||
) -> Widget | None:
|
||||
assert session is not None, "Session must be provided by with_session decorator"
|
||||
widget = await session.get(Widget, widget_id)
|
||||
if widget is None:
|
||||
return None
|
||||
|
||||
await session.delete(widget)
|
||||
await session.flush()
|
||||
return widget
|
||||
```
|
||||
|
||||
Update and delete load the row through the same session that mutates it. This avoids accepting detached instances from an earlier standalone read and gives both operations an explicit `None` result that the application layer can map to a domain or HTTP error. Delete returns the loaded object for callers that need its values, but that object represents a row scheduled for deletion and must not be reused as persistent state. List operations validate their bounds and order by the primary key so pagination is deterministic. Add a unique tiebreaker whenever ordering by a non-unique field.
|
||||
|
||||
`flush()` sends pending writes and populates ordinary generated primary keys. It does not itself commit. A decorated write called without a session will therefore roll back when its owned session closes unless the function explicitly commits. Prefer passing a transaction-scoped session so several writes compose atomically. Use `await session.refresh(widget)` only when the operation deliberately needs database-generated state that was not returned during the flush; an unconditional refresh adds another query.
|
||||
|
||||
---
|
||||
|
||||
## Repository Object
|
||||
|
||||
A repository can provide a stable domain-facing interface when several callers need the same grouped operations. It remains stateless here: every method requires a session and delegates to the analogous function.
|
||||
|
||||
```python
|
||||
from sqlmodel.ext.asyncio.session import AsyncSession
|
||||
|
||||
class WidgetRepository:
|
||||
async def create(
|
||||
self,
|
||||
session: AsyncSession,
|
||||
name: str,
|
||||
description: str | None = None,
|
||||
) -> Widget:
|
||||
return await create_widget(
|
||||
name,
|
||||
description,
|
||||
session=session,
|
||||
)
|
||||
|
||||
async def get(
|
||||
self,
|
||||
session: AsyncSession,
|
||||
widget_id: int,
|
||||
) -> Widget | None:
|
||||
return await get_widget(widget_id, session=session)
|
||||
|
||||
async def list(
|
||||
self,
|
||||
session: AsyncSession,
|
||||
*,
|
||||
offset: int = 0,
|
||||
limit: int = 100,
|
||||
) -> list[Widget]:
|
||||
return await list_widgets(
|
||||
offset=offset,
|
||||
limit=limit,
|
||||
session=session,
|
||||
)
|
||||
|
||||
async def update(
|
||||
self,
|
||||
session: AsyncSession,
|
||||
widget_id: int,
|
||||
name: str,
|
||||
description: str | None,
|
||||
) -> Widget | None:
|
||||
return await update_widget(
|
||||
widget_id,
|
||||
name,
|
||||
description,
|
||||
session=session,
|
||||
)
|
||||
|
||||
async def delete(
|
||||
self,
|
||||
session: AsyncSession,
|
||||
widget_id: int,
|
||||
) -> Widget | None:
|
||||
return await delete_widget(widget_id, session=session)
|
||||
```
|
||||
|
||||
The object is intentionally thin. Tests pass a transaction-scoped test session directly. The caller always owns that session and its transaction, and the repository never closes or commits it.
|
||||
|
||||
If a read participates in a later write, pass the same session and place both operations inside the explicit transaction. This avoids splitting one use case across sessions and keeps SQLAlchemy's autobegin behavior from obscuring transaction ownership. Add a repository only when its naming, shared query policy, dependency substitution, or domain boundary improves the application. Independent functions remain a valid and often clearer design.
|
||||
|
||||
---
|
||||
|
||||
## Transaction Ownership
|
||||
|
||||
Compose multiple calls under one use-case transaction. `db_transaction_scope()` owns the standalone engine, factory, session, and transaction lifetimes. Decorated CRUD functions detect the supplied session and borrow it; repository methods receive it directly.
|
||||
|
||||
```python
|
||||
from .session import db_transaction_scope
|
||||
|
||||
|
||||
async def replace_widget(
|
||||
repository: WidgetRepository,
|
||||
widget_id: int,
|
||||
replacement_name: str,
|
||||
replacement_description: str | None = None,
|
||||
) -> Widget | None:
|
||||
async with db_transaction_scope() as active_session:
|
||||
deleted_widget = await repository.delete(
|
||||
active_session,
|
||||
widget_id,
|
||||
)
|
||||
if deleted_widget is None:
|
||||
return None
|
||||
|
||||
return await repository.create(
|
||||
active_session,
|
||||
replacement_name,
|
||||
replacement_description,
|
||||
)
|
||||
```
|
||||
|
||||
If creation fails, deletion rolls back with it. Inside an already-running application, prefer `async with session_factory.begin()` or `async with session.begin()` over `db_transaction_scope()` so the application-owned engine and factory remain in use. Do not add direct `commit()` calls to CRUD functions or repository methods because that prevents callers from composing several operations atomically. See [transaction boundaries](transactions.md) and [session management](session.md) for ownership details.
|
||||
|
||||
---
|
||||
|
||||
## Anti-Patterns
|
||||
|
||||
- Storing one mutable `AsyncSession` on a long-lived repository object.
|
||||
- Creating sessions manually inside functions already using `@with_session`.
|
||||
- Passing database configuration through every CRUD call instead of injecting a session at the data-access boundary.
|
||||
- Assuming decorator-owned write sessions commit on close.
|
||||
- Forwarding explicit `session=None` when decorator injection was intended.
|
||||
- Accepting unbounded list queries.
|
||||
- Accepting detached ORM instances for update or delete when an identifier can be resolved in the active session.
|
||||
- Accessing unloaded attributes after a standalone repository read has closed its owned session.
|
||||
|
||||
---
|
||||
|
||||
## Operational Checks
|
||||
|
||||
- Every CRUD call receives a task-local `AsyncSession`.
|
||||
- Standalone reads create and close a session at the service or application boundary.
|
||||
- Standalone reads may omit `session`; decorated writes receive a transaction-scoped session or explicitly own their commit policy.
|
||||
- Supplied write sessions remain caller-owned.
|
||||
- Each complete write operation declares a visible transaction boundary.
|
||||
- List operations have pagination and deterministic ordering where required.
|
||||
- Update requires values for both mutable fields; passing `None` explicitly clears the nullable description.
|
||||
- Get, update, and delete use the same identifier and missing-row semantics.
|
||||
- Decorated functions accept an optional keyword-only session; repository methods require one explicitly.
|
||||
- Standalone service reads load all state needed after their owned session closes.
|
||||
- Repository objects hold query policy when useful, never database configuration or request-scoped session state.
|
||||
|
||||
---
|
||||
|
||||
## Testing Checks
|
||||
|
||||
- Create tests verify generated identifiers and persisted field values after commit.
|
||||
- Get and list tests cover found, missing, pagination, and ordering behavior.
|
||||
- List tests reject negative offsets and limits outside the supported range.
|
||||
- Update tests cover replacement of both mutable fields, including clearing the nullable description.
|
||||
- Update and delete tests cover missing identifiers without mutating the database.
|
||||
- Delete tests verify the returned row and its absence after commit.
|
||||
- Failure tests verify that a surrounding transaction rolls back all composed CRUD calls.
|
||||
- Optional-session read tests verify borrowed sessions remain open and owned sessions close without committing.
|
||||
- Decorated write tests verify supplied transactions remain caller-owned and omitted sessions do not imply a commit.
|
||||
- Composition tests pass one active session through several CRUD calls and verify one atomic commit or rollback.
|
||||
@@ -0,0 +1,281 @@
|
||||
# Async SQLAlchemy Engine
|
||||
|
||||
!!! info "Primary sources"
|
||||
- [Python `asynccontextmanager`](https://docs.python.org/3/library/contextlib.html#contextlib.asynccontextmanager)
|
||||
- [Python `functools.cache`](https://docs.python.org/3/library/functools.html#functools.cache)
|
||||
- [SQLAlchemy connections](https://docs.sqlalchemy.org/en/21/core/connections.html)
|
||||
- [SQLAlchemy asyncio extension](https://docs.sqlalchemy.org/en/21/orm/extensions/asyncio.html)
|
||||
- [SQLAlchemy pooling and multiprocessing](https://docs.sqlalchemy.org/en/21/core/pooling.html#pooling-multiprocessing)
|
||||
- [SQLAlchemy SQLite transaction control](https://docs.sqlalchemy.org/en/21/dialects/sqlite.html#enabling-non-legacy-sqlite-transactional-modes-with-the-sqlite3-or-aiosqlite-driver)
|
||||
- [SQLAlchemy SQLite foreign-key support](https://docs.sqlalchemy.org/en/21/dialects/sqlite.html#foreign-key-support)
|
||||
- [SQLite PRAGMA reference](https://www.sqlite.org/pragma.html)
|
||||
- [`nicegui-db` engine implementation](https://forgejo.john-stream.com/john/nicegui-db/src/commit/126bc26ad8635a86bacf684d7bda409230347597/src/nicegui_db/db/engine.py)
|
||||
|
||||
---
|
||||
|
||||
## Engine Ownership Model
|
||||
|
||||
Resolve one async engine for each database URL within an application, worker, command, or test lifecycle.
|
||||
|
||||
- SQLAlchemy guidance: the engine is intended as a long-lived, concurrent registry over pooled DB connections, not a per-operation object.
|
||||
- `get_engine(database_url)` owns URL-keyed engine construction and caching.
|
||||
- The composition root enters `engine_scope(database_url)` once and therefore owns initialization and disposal.
|
||||
- Services and repositories receive a session or session factory; they do not resolve an engine.
|
||||
|
||||
!!! tip "Practical rule"
|
||||
- Exactly one cached engine for each database URL during an active application-owned lifecycle.
|
||||
- Exactly one active owning `engine_scope()` for a given URL.
|
||||
- Zero `create_async_engine(...)` calls in feature code.
|
||||
- Zero engine lookup or disposal calls in repository code.
|
||||
|
||||
---
|
||||
|
||||
## Cached Engine Resolution
|
||||
|
||||
[`functools.cache`](https://docs.python.org/3/library/functools.html#functools.cache) makes the database URL the engine identity:
|
||||
|
||||
```python
|
||||
from functools import cache
|
||||
|
||||
from sqlalchemy.ext.asyncio import AsyncEngine
|
||||
from sqlalchemy.ext.asyncio import create_async_engine
|
||||
|
||||
|
||||
@cache
|
||||
def get_engine(database_url: str) -> AsyncEngine:
|
||||
engine = create_async_engine(database_url, pool_pre_ping=True)
|
||||
if engine.dialect.name == "sqlite":
|
||||
configure_aiosqlite_engine(engine)
|
||||
return engine
|
||||
```
|
||||
|
||||
Repeated calls with the same exact URL return the same `AsyncEngine`; different URLs produce independent cache entries. Construction configures the dialect and pool but normally does not open a database connection until the first operation. SQLite event listeners are installed only when a new cached engine is constructed, before its first connection.
|
||||
|
||||
Resolve settings into the final URL before calling `get_engine()`. Services and repositories should not call it directly: the cache controls construction identity, not ownership.
|
||||
|
||||
## Owning Engine Scope
|
||||
|
||||
Use one [`asynccontextmanager`](https://docs.python.org/3/library/contextlib.html#contextlib.asynccontextmanager) to pair cached resolution and optional schema initialization with disposal:
|
||||
|
||||
```python
|
||||
from collections.abc import AsyncGenerator
|
||||
from contextlib import asynccontextmanager
|
||||
|
||||
from sqlalchemy.ext.asyncio import AsyncEngine
|
||||
from sqlmodel import SQLModel
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def engine_scope(
|
||||
database_url: str,
|
||||
*,
|
||||
initialize: bool = True,
|
||||
) -> AsyncGenerator[AsyncEngine]:
|
||||
engine = get_engine(database_url)
|
||||
if initialize:
|
||||
await initialize_db(database_url)
|
||||
|
||||
try:
|
||||
yield engine
|
||||
finally:
|
||||
await dispose_engine(database_url)
|
||||
|
||||
|
||||
async def initialize_db(database_url: str) -> None:
|
||||
from . import models # noqa: F401
|
||||
|
||||
engine = get_engine(database_url)
|
||||
async with engine.begin() as connection:
|
||||
await connection.run_sync(SQLModel.metadata.create_all)
|
||||
|
||||
|
||||
async def dispose_engine(database_url: str) -> None:
|
||||
engine = get_engine(database_url)
|
||||
try:
|
||||
await engine.dispose()
|
||||
finally:
|
||||
get_engine.cache_clear()
|
||||
```
|
||||
|
||||
The code that enters `engine_scope()` owns the engine. It keeps that scope open for the complete application, worker, command, or test lifecycle and passes the yielded engine into session-factory construction. Successful and exceptional exits both dispose the pool and invalidate cached engine resolution.
|
||||
|
||||
Initialization imports the model package so every table is registered, then runs `SQLModel.metadata.create_all()` in `engine.begin()`. This is suitable for the template and focused tests. Use migrations instead when schema evolution is part of the deployment contract. Pass `initialize=False` only when another owner provisions the schema or a test is directly exercising construction without schema setup.
|
||||
|
||||
`dispose_engine()` clears the complete function cache, not only the requested URL. This matches the template and is safe under its intended single-database lifecycle. Applications that own several simultaneously active database URLs need per-key lifecycle management rather than this global invalidation behavior.
|
||||
|
||||
Workers, scripts, and other composition roots enter `database_scope()` directly:
|
||||
|
||||
```python
|
||||
async with database_scope(settings.database_url) as session_factory:
|
||||
await run_worker(session_factory)
|
||||
```
|
||||
|
||||
`database_scope()` is defined in [session management](session.md). It enters `engine_scope()` and creates the factory bound to the yielded engine.
|
||||
|
||||
Do not overlap two owning scopes for the same URL. Both resolve the same cached engine, and the first scope to exit disposes it and clears the cache while the other still refers to it. For several fixed databases, use one non-overlapping owner per URL and account for global cache invalidation; use [`AsyncExitStack`](https://docs.python.org/3/library/contextlib.html#contextlib.AsyncExitStack) only after adopting lifecycle semantics that support several simultaneous owners.
|
||||
|
||||
When directly testing engine construction or lifecycle behavior, enter `engine_scope()` in the test or fixture. Exiting the context disposes the engine even when the test fails and clears the cache for the next lifecycle.
|
||||
|
||||
See [FastAPI database integration](fastapi.md) for adapting `database_scope()` to application lifespan and dependency injection.
|
||||
|
||||
---
|
||||
|
||||
## Driver URLs (Project Requirement: asyncpg + aiosqlite)
|
||||
|
||||
Use SQLAlchemy async driver URLs:
|
||||
|
||||
- PostgreSQL: `postgresql+asyncpg://user:pass@host:5432/dbname`
|
||||
- SQLite: `sqlite+aiosqlite:///./app.db`
|
||||
|
||||
!!! warning "Driver compatibility"
|
||||
- Do not mix sync drivers, for example `psycopg2`, with `create_async_engine()`.
|
||||
- Keep URL construction centralized in settings/config, not in feature modules.
|
||||
|
||||
---
|
||||
|
||||
## SQLite Connection and Transaction Policy
|
||||
|
||||
SQLite settings do not form one indivisible bundle:
|
||||
|
||||
- `PRAGMA foreign_keys=ON` is a correctness requirement when the schema declares foreign keys. SQLite requires it on every connection, including the connection used by `metadata.create_all()`.
|
||||
- Disabling the driver's implicit `BEGIN` and emitting `BEGIN` from SQLAlchemy provides non-legacy transaction behavior for `aiosqlite`. This makes SELECT, DDL, and SAVEPOINT behavior participate in SQLAlchemy's transaction boundary consistently.
|
||||
- `PRAGMA busy_timeout` is a per-connection lock-wait policy. Choose the duration from the application's latency and contention requirements.
|
||||
- `PRAGMA journal_mode=WAL` is an optional file-database concurrency policy. WAL persists in the database file, cannot be enabled for an in-memory database, and is not a substitute for transaction control.
|
||||
|
||||
Install instance-level listeners exactly once, immediately after constructing an `aiosqlite` engine and before its first connection:
|
||||
|
||||
```python
|
||||
from sqlalchemy import event
|
||||
from sqlalchemy.engine import Connection
|
||||
from sqlalchemy.engine.interfaces import DBAPIConnection
|
||||
from sqlalchemy.ext.asyncio import AsyncEngine
|
||||
|
||||
|
||||
def configure_aiosqlite_engine(
|
||||
engine: AsyncEngine,
|
||||
*,
|
||||
busy_timeout_ms: int | None = 30_000,
|
||||
enable_wal: bool = False,
|
||||
) -> None:
|
||||
if engine.dialect.name != "sqlite" or engine.dialect.driver != "aiosqlite":
|
||||
raise ValueError("Expected a sqlite+aiosqlite engine")
|
||||
if busy_timeout_ms is not None and busy_timeout_ms < 0:
|
||||
raise ValueError("busy_timeout_ms must be non-negative")
|
||||
|
||||
@event.listens_for(engine.sync_engine, "connect")
|
||||
def configure_connection(dbapi_connection: DBAPIConnection, _: object) -> None:
|
||||
dbapi_connection.isolation_level = None
|
||||
cursor = dbapi_connection.cursor()
|
||||
try:
|
||||
cursor.execute("PRAGMA foreign_keys=ON")
|
||||
if busy_timeout_ms is not None:
|
||||
cursor.execute(f"PRAGMA busy_timeout={busy_timeout_ms}")
|
||||
if enable_wal:
|
||||
cursor.execute("PRAGMA journal_mode=WAL")
|
||||
journal_mode = cursor.fetchone()
|
||||
if journal_mode is None or journal_mode[0].lower() != "wal":
|
||||
raise RuntimeError("SQLite could not enable WAL mode")
|
||||
finally:
|
||||
cursor.close()
|
||||
|
||||
@event.listens_for(engine.sync_engine, "begin")
|
||||
def begin_transaction(connection: Connection) -> None:
|
||||
connection.exec_driver_sql("BEGIN")
|
||||
```
|
||||
|
||||
The `connect` listener receives the adapted synchronous DBAPI connection exposed by `engine.sync_engine`; event callbacks themselves are synchronous even though application queries use the async engine. Setting `isolation_level=None` and adding the `begin` listener are one transaction-control strategy and must remain paired. Do not combine this pair with SQLAlchemy's driver-level `AUTOCOMMIT` isolation mode.
|
||||
|
||||
The default above enables foreign keys and modern transaction boundaries for file and in-memory databases. Enable WAL only for a file-backed database after deciding that its read/write concurrency model is appropriate. Treat `30_000` as an example policy, not a universal default; `connect_args={"timeout": 30.0}` at engine construction is another way to configure the underlying SQLite lock timeout.
|
||||
|
||||
---
|
||||
|
||||
## Pooling Defaults and Tuning
|
||||
|
||||
Default behavior is usually correct first:
|
||||
|
||||
- Async engines use async-compatible pooling (`AsyncAdaptedQueuePool`) by default.
|
||||
- Start with defaults, then tune from observed load (`pool_size`, `max_overflow`, `pool_timeout`, `pool_recycle`).
|
||||
- Enable `pool_pre_ping=True` for safer stale-connection handling in long-running services.
|
||||
|
||||
When to switch pool strategy:
|
||||
|
||||
- `NullPool` if you explicitly need no pooling (special environments, some tests, or strict cross-loop constraints).
|
||||
- Keep in mind this increases connect/disconnect churn.
|
||||
|
||||
### When `StaticPool` Is Appropriate
|
||||
|
||||
Use [`StaticPool`](https://docs.sqlalchemy.org/en/21/core/pooling.html#sqlalchemy.pool.StaticPool) only when every checkout must reuse one DBAPI connection and all database access is serialized. Typical cases are:
|
||||
|
||||
- A serial test suite using a private in-memory SQLite database. The `sqlite+aiosqlite://` URL already selects `StaticPool` automatically, so specifying `poolclass=StaticPool` is normally redundant.
|
||||
- A narrowly scoped SQLite engine that must preserve connection-local state, such as temporary tables, across SQLAlchemy connection or session checkouts.
|
||||
|
||||
When explicit configuration is required:
|
||||
|
||||
```python
|
||||
from sqlalchemy.ext.asyncio import create_async_engine
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
engine = create_async_engine(
|
||||
"sqlite+aiosqlite:///./test.db",
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
```
|
||||
|
||||
`StaticPool` is not a general performance optimization or a way to make SQLite concurrent. All sessions share one underlying connection and its single transaction state, so one session's `COMMIT` or `ROLLBACK` can interfere with another session. Do not use it when several sessions or tasks may access the engine concurrently. For concurrent in-memory work, use a named shared-cache SQLite URL so pooled connections have independent transaction state, or use a temporary file database. See [SQLite test targets](testing.md#sqlite-targets) for those patterns.
|
||||
|
||||
---
|
||||
|
||||
## Disposal Semantics
|
||||
|
||||
`dispose_engine(database_url)` resolves the cached engine, awaits `engine.dispose()`, and clears the engine cache in a `finally` block. `engine.dispose()` replaces/disposes the pool, but only checked-in connections are immediately closed.
|
||||
|
||||
Rules:
|
||||
- Dispose when the app is shutting down.
|
||||
- Clear cached resolution even when disposal raises, so a later lifecycle cannot receive the failed engine object.
|
||||
- Dispose before reusing an engine across event loops.
|
||||
- In forked child-process initialization, use `engine.dispose(close=False)` (sync API guidance) so child processes do not touch parent-held connections.
|
||||
|
||||
Avoid relying on garbage collection for engine cleanup in async code.
|
||||
|
||||
---
|
||||
|
||||
## Event Loop and Process Boundaries
|
||||
|
||||
Do not share pooled connections across boundaries:
|
||||
|
||||
- Multiple event loops: do not reuse the same pooled async engine across loops unless you intentionally disable pooling (`NullPool`) or dispose before handoff.
|
||||
- Multiprocessing/fork: pooled connections must not be inherited for active use across process boundaries.
|
||||
|
||||
This prevents broken socket state and cross-process connection corruption.
|
||||
|
||||
---
|
||||
|
||||
## What Not to Do
|
||||
|
||||
- Create an engine inside each operation or unit of work.
|
||||
- Create/dispose engines inside repository methods.
|
||||
- Resolve an engine from repositories instead of injecting a session dependency.
|
||||
- Keep engine creation as a hidden side effect of import-time module globals.
|
||||
- Keep a session factory alive after its bound engine scope exits.
|
||||
- Enter overlapping engine scopes for the same cached URL.
|
||||
- Treat `cache_clear()` as per-URL invalidation when it clears every cached engine.
|
||||
- Use `metadata.create_all()` as a substitute for required production migrations.
|
||||
- Install the same SQLite event listeners more than once on one engine.
|
||||
- Enable WAL blindly for in-memory SQLite or treat a busy timeout as a concurrency guarantee.
|
||||
|
||||
---
|
||||
|
||||
## Engine Design Checklist
|
||||
|
||||
- One cached engine per exact database URL during an active lifecycle.
|
||||
- One owning engine scope per URL, with no overlapping owners.
|
||||
- Cached resolution, optional initialization, disposal, and cache invalidation follow one framework-independent lifecycle.
|
||||
- The composition root enters the database scope once and keeps it open until shutdown.
|
||||
- Session factory created inside, and never outlives, its engine scope.
|
||||
- Model registration occurs before `metadata.create_all()` when initialization is enabled.
|
||||
- Async driver URL matches backend (`asyncpg` or `aiosqlite`).
|
||||
- `aiosqlite` foreign-key and transaction listeners installed once before first use.
|
||||
- WAL enabled only as an explicit policy for a file-backed SQLite database.
|
||||
- Pooling strategy is explicit for non-default needs.
|
||||
- No feature-path engine creation.
|
||||
- Tests enter the same scope and receive deterministic disposal plus cache cleanup.
|
||||
@@ -0,0 +1,192 @@
|
||||
# FastAPI Database Integration
|
||||
|
||||
!!! info "Primary sources"
|
||||
- [FastAPI lifespan events](https://fastapi.tiangolo.com/advanced/events/)
|
||||
- [FastAPI dependencies with `yield`](https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-with-yield/)
|
||||
- [FastAPI dependency overrides](https://fastapi.tiangolo.com/advanced/testing-dependencies/)
|
||||
- [SQLAlchemy asyncio extension](https://docs.sqlalchemy.org/en/21/orm/extensions/asyncio.html)
|
||||
- [`nicegui-db` application lifespan](https://forgejo.john-stream.com/john/nicegui-db/src/commit/126bc26ad8635a86bacf684d7bda409230347597/src/nicegui_db/ui/app.py)
|
||||
- [`nicegui-db` database dependencies](https://forgejo.john-stream.com/john/nicegui-db/src/commit/126bc26ad8635a86bacf684d7bda409230347597/src/nicegui_db/ui/dependency.py)
|
||||
|
||||
---
|
||||
|
||||
## Purpose
|
||||
|
||||
Connect the framework-independent database tools to FastAPI:
|
||||
|
||||
- lifespan enters one application-owned `database_scope()`,
|
||||
- application state holds settings and the resulting session factory,
|
||||
- dependencies create one session per request,
|
||||
- `Annotated` aliases make route ownership concise and explicit.
|
||||
|
||||
The underlying resource and transaction rules remain in [engine lifecycle](engine.md), [session management](session.md), and [transaction boundaries](transactions.md).
|
||||
|
||||
---
|
||||
|
||||
## Lifespan Ownership
|
||||
|
||||
Enter `database_scope()` once for the complete application lifecycle. Store the session factory, not the engine, because request code needs sessions rather than direct pool access:
|
||||
|
||||
```python
|
||||
from collections.abc import AsyncGenerator
|
||||
from contextlib import asynccontextmanager
|
||||
|
||||
from fastapi import FastAPI
|
||||
|
||||
from .config import Settings
|
||||
from .config import get_database_url
|
||||
from .db import database_scope
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(settings: Settings, app: FastAPI) -> AsyncGenerator[None]:
|
||||
app.state.settings = settings
|
||||
db_url = get_database_url(settings)
|
||||
|
||||
try:
|
||||
async with database_scope(db_url) as session_factory:
|
||||
app.state.session_factory = session_factory
|
||||
yield
|
||||
finally:
|
||||
del app.state.settings
|
||||
del app.state.session_factory
|
||||
```
|
||||
|
||||
The application factory binds `settings` to lifespan, for example with `partial(lifespan, settings)`. Lifespan does not construct resources per request. It enters the same framework-independent scope used by scripts, workers, and tests, keeps that scope open while requests are served, and lets it dispose the engine and clear cached engine resolution during shutdown.
|
||||
|
||||
The template's unconditional `del app.state.session_factory` mirrors an expected successful startup. If `database_scope()` raises before assignment, cleanup can raise `AttributeError` and obscure the startup error. A production hardening option is to assign a sentinel before the `try` or delete conditionally; that changes failure behavior and is not part of the exact template mechanics.
|
||||
|
||||
Only store the engine too when application-level code genuinely needs direct Core operations, pool instrumentation, or engine-specific diagnostics. Routes and repositories should normally receive an `AsyncSession`.
|
||||
|
||||
---
|
||||
|
||||
## Session Factory Dependency
|
||||
|
||||
A synchronous dependency retrieves the already-created factory from application state:
|
||||
|
||||
```python
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import Depends
|
||||
from fastapi import Request
|
||||
|
||||
from .session import SessionFactory
|
||||
|
||||
|
||||
def _get_session_factory(request: Request) -> SessionFactory:
|
||||
return request.app.state.session_factory
|
||||
|
||||
|
||||
type SessionFactoryDep = Annotated[SessionFactory, Depends(_get_session_factory)]
|
||||
```
|
||||
|
||||
`Depends()` does not create or cache a factory here. It only exposes the lifespan-owned object. This function is also the narrow seam that tests can override when they need a different factory.
|
||||
|
||||
---
|
||||
|
||||
## Request Session Dependencies
|
||||
|
||||
Use a session-only dependency for reads and other request conversations that must not commit implicitly:
|
||||
|
||||
```python
|
||||
from collections.abc import AsyncGenerator
|
||||
|
||||
from sqlmodel.ext.asyncio.session import AsyncSession
|
||||
|
||||
|
||||
async def _get_session(session_factory: SessionFactoryDep) -> AsyncGenerator[AsyncSession]:
|
||||
async with session_factory() as owned_session:
|
||||
yield owned_session
|
||||
|
||||
|
||||
type SessionDep = Annotated[AsyncSession, Depends(_get_session)]
|
||||
```
|
||||
|
||||
The dependency creates and closes one session per request. Closing rolls back any unfinished autobegun transaction; it does not commit.
|
||||
|
||||
---
|
||||
|
||||
## Route Usage
|
||||
|
||||
Read route:
|
||||
|
||||
```python
|
||||
@router.get("/items/{item_id}")
|
||||
async def get_item(item_id: int, session: SessionDep) -> Item | None:
|
||||
return await find_item(session, item_id)
|
||||
```
|
||||
|
||||
Write route:
|
||||
|
||||
```python
|
||||
@router.post("/items")
|
||||
async def create_item(payload: ItemCreate, session: SessionDep) -> Item:
|
||||
async with session.begin():
|
||||
return await insert_item(session, payload)
|
||||
```
|
||||
|
||||
The template exposes only `SessionDep`; it does not hide commit behavior in dependency teardown. Choose one visible write convention per application:
|
||||
|
||||
- place `async with session.begin():` around a complete write unit, which commits on success and rolls back on exception; or
|
||||
- call `await session.commit()` explicitly after all writes when the route is the complete unit, as the template's simple UI action does.
|
||||
|
||||
The context-manager form scales better to several statements and makes exception rollback visible. Direct `commit()` is concise but requires the route to preserve the single-commit invariant and handle any recovery needs. Do not combine both conventions in one route. Lower-level data-access functions receive the existing session and remain unaware of FastAPI.
|
||||
|
||||
---
|
||||
|
||||
## Background Work
|
||||
|
||||
A request session belongs to that request and must not be retained by a background task. Inject or otherwise provide the application session factory, then create a new session inside the task:
|
||||
|
||||
```python
|
||||
async def run_background_job(session_factory: SessionFactory) -> None:
|
||||
async with session_factory.begin() as session:
|
||||
await process_pending_items(session)
|
||||
```
|
||||
|
||||
If work must survive application shutdown, it needs an independently owned worker lifecycle rather than the FastAPI lifespan-owned factory.
|
||||
|
||||
---
|
||||
|
||||
## Testing and Overrides
|
||||
|
||||
Override the narrow dependency that matches the test objective:
|
||||
|
||||
- Override `_get_session_factory` to preserve production request-session behavior with a test factory.
|
||||
- Override `_get_session` when a test must inject one transaction-scoped session directly.
|
||||
- Verify each lifespan receives a fresh engine and session factory and removes application state during teardown.
|
||||
- Remove overrides during teardown so mutable application state does not leak between tests.
|
||||
|
||||
```python
|
||||
app.dependency_overrides[_get_session] = get_test_session
|
||||
try:
|
||||
yield app
|
||||
finally:
|
||||
app.dependency_overrides.pop(_get_session, None)
|
||||
```
|
||||
|
||||
See [database testing](testing.md) for outer transactions, SAVEPOINT-backed fixtures, and database target selection.
|
||||
|
||||
---
|
||||
|
||||
## Anti-Patterns
|
||||
|
||||
- Creating an engine or session factory in a request dependency.
|
||||
- Reading settings and constructing database resources from repositories.
|
||||
- Storing one mutable `AsyncSession` on `app.state`.
|
||||
- Sharing a request session with concurrent or background tasks.
|
||||
- Assuming `SessionDep` commits when dependency cleanup runs.
|
||||
- Keeping `app.state.session_factory` after its `database_scope()` exits.
|
||||
- Using deprecated startup and shutdown event handlers alongside lifespan.
|
||||
|
||||
---
|
||||
|
||||
## Integration Checklist
|
||||
|
||||
- Lifespan enters exactly one `database_scope()` for each application lifecycle.
|
||||
- Application state stores the yielded session factory.
|
||||
- Session dependencies create and close one session per request.
|
||||
- The session dependency owns request session closure but not commit behavior.
|
||||
- Routes use `Annotated` aliases and receive sessions, not engines.
|
||||
- Background tasks create their own sessions from a still-live factory.
|
||||
- Tests override and restore dependencies deterministically.
|
||||
@@ -0,0 +1,101 @@
|
||||
# Preventing Implicit ORM I/O (Asyncio)
|
||||
|
||||
!!! info "Primary sources"
|
||||
- [Preventing implicit I/O with AsyncSession](https://docs.sqlalchemy.org/en/21/orm/extensions/asyncio.html#preventing-implicit-io-when-using-asyncsession)
|
||||
- [SQLAlchemy relationship loading](https://docs.sqlalchemy.org/en/21/orm/queryguide/relationships.html)
|
||||
|
||||
??? abstract "Decision metadata"
|
||||
- Status: adopted
|
||||
- Decision level: advisory
|
||||
- Applies to: api-runtime, workers, tests
|
||||
- Last reviewed: 2026-06-17
|
||||
|
||||
---
|
||||
|
||||
## Purpose
|
||||
|
||||
Minimize unexpected database round-trips caused by attribute access in async ORM code.
|
||||
|
||||
In asyncio applications, hidden lazy loads are easy to miss and can produce runtime surprises. This guide defines explicit-loading defaults and progressive enforcement practices.
|
||||
|
||||
---
|
||||
|
||||
## Scope and Non-Goals
|
||||
|
||||
- In scope: relationship loading strategy, post-commit attribute access, explicit refresh/awaitable access patterns.
|
||||
- Out of scope: full ORM performance tuning and domain-specific query architecture.
|
||||
|
||||
---
|
||||
|
||||
## Rules
|
||||
|
||||
- Prefer explicit eager loading for data required by endpoint/service outputs.
|
||||
- Avoid relying on implicit lazy-load behavior in request critical paths.
|
||||
- Keep `expire_on_commit=False` unless strict expiration behavior is intentionally required.
|
||||
- Use explicit refresh or awaitable-attribute access when loading deferred state is necessary.
|
||||
|
||||
---
|
||||
|
||||
## Recommended Patterns
|
||||
|
||||
### Pattern A: Eager-load what you need
|
||||
|
||||
```python
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
stmt = select(User).options(selectinload(User.roles))
|
||||
users = (await session.scalars(stmt)).all()
|
||||
```
|
||||
|
||||
### Pattern B: Explicit refresh of named attributes
|
||||
|
||||
```python
|
||||
user = await session.get(User, user_id)
|
||||
await session.refresh(user, ["roles"])
|
||||
```
|
||||
|
||||
### Pattern C: Awaitable attribute access where needed
|
||||
|
||||
```python
|
||||
# Requires AsyncAttrs mixin on mapped base or class.
|
||||
roles = await user.awaitable_attrs.roles
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Practical Enforcement Model
|
||||
|
||||
Require explicit I/O behavior on every async ORM path:
|
||||
|
||||
1. Define loader options for relationships and deferred columns needed by the operation.
|
||||
2. Use `refresh()` or awaitable attributes only when the additional query is deliberate and visible.
|
||||
3. Add review checks that reject unplanned lazy-load paths.
|
||||
|
||||
This keeps event-loop behavior predictable and makes query boundaries reviewable from the code.
|
||||
|
||||
---
|
||||
|
||||
## Anti-Patterns
|
||||
|
||||
- Returning ORM objects from handlers and triggering lazy loads during serialization.
|
||||
- Assuming post-commit attribute access will always be loaded without explicit strategy.
|
||||
- Relying on broad expiration + implicit reload behavior in async request flows.
|
||||
- Enabling relationship patterns that hide SQL behavior in critical code paths.
|
||||
|
||||
---
|
||||
|
||||
## Operational Checks
|
||||
|
||||
- Endpoint query blocks define loader options for returned related data.
|
||||
- Critical handlers do not depend on incidental lazy loads.
|
||||
- Known exceptions are documented with rationale and follow-up items.
|
||||
|
||||
---
|
||||
|
||||
## Testing Checks
|
||||
|
||||
- Integration tests cover endpoints that return related objects.
|
||||
- Tests verify expected data is present without hidden secondary query surprises.
|
||||
- Regression tests exist for routes previously affected by implicit-load failures.
|
||||
|
||||
@@ -0,0 +1,36 @@
|
||||
# FastAPI Async SQLAlchemy References Index
|
||||
|
||||
Purpose: concept registry for the principles, mechanics, and implementation guidance used by this skill.
|
||||
|
||||
---
|
||||
|
||||
## Concepts
|
||||
|
||||
| Concept | File | Status | Decision Level | Owner | Last Reviewed |
|
||||
|---|---|---|---|---|---|
|
||||
| Engine lifecycle and ownership | [engine.md](engine.md) | adopted | mandatory | platform/backend | 2026-08-06 |
|
||||
| Session factory and scope | [session.md](session.md) | adopted | mandatory | platform/backend | 2026-08-06 |
|
||||
| FastAPI lifespan and dependency injection | [fastapi.md](fastapi.md) | adopted | mandatory | platform/backend | 2026-08-06 |
|
||||
| Transaction boundaries | [transactions.md](transactions.md) | adopted | mandatory | platform/backend | 2026-06-17 |
|
||||
| Implicit ORM I/O under asyncio | [implicit_io.md](implicit_io.md) | adopted | advisory | platform/backend | 2026-06-17 |
|
||||
| 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-08-06 |
|
||||
| Basic CRUD repository and functions | [crud.md](crud.md) | adopted | advisory | platform/backend | 2026-08-06 |
|
||||
| Test database targets and fixture data | [testing.md](testing.md) | adopted | mandatory | platform/backend | 2026-08-06 |
|
||||
|
||||
---
|
||||
|
||||
## How to Use This Folder
|
||||
|
||||
- `SKILL.md` defines the explanatory workflow and shared mental model.
|
||||
- Each concept doc defines policy-level guidance for one concern.
|
||||
- Use the template in [template.md](template.md) for new concept docs.
|
||||
- Keep references source-linked and implementation snippets minimal.
|
||||
|
||||
---
|
||||
|
||||
## Update Rules
|
||||
|
||||
- If a PR changes database lifecycle/session/ORM loading behavior, update the relevant concept file.
|
||||
- Keep `Status`, `Decision Level`, and `Last Reviewed` current.
|
||||
- Use `advisory` for recommendations that depend on application context; use `mandatory` for required runtime policy.
|
||||
@@ -0,0 +1,107 @@
|
||||
# DB Observability and Resilience
|
||||
|
||||
!!! info "Primary sources"
|
||||
- [SQLAlchemy pooling](https://docs.sqlalchemy.org/en/21/core/pooling.html)
|
||||
- [SQLAlchemy engine configuration](https://docs.sqlalchemy.org/en/21/core/engines.html)
|
||||
- [SQLAlchemy events](https://docs.sqlalchemy.org/en/21/core/events.html)
|
||||
- [FastAPI lifespan events](https://fastapi.tiangolo.com/advanced/events/)
|
||||
|
||||
??? abstract "Decision metadata"
|
||||
- Status: adopted
|
||||
- Decision level: mandatory
|
||||
- Applies to: api-runtime, workers, tests
|
||||
- Last reviewed: 2026-06-17
|
||||
|
||||
---
|
||||
|
||||
## Purpose
|
||||
|
||||
Define baseline observability and resilience practices for DB connectivity in async FastAPI + SQLAlchemy apps.
|
||||
|
||||
Goals:
|
||||
|
||||
- detect and recover from stale/disconnected connections,
|
||||
- expose useful diagnostics for pool/engine behavior,
|
||||
- make readiness/liveness signals meaningful.
|
||||
|
||||
---
|
||||
|
||||
## Scope and Non-Goals
|
||||
|
||||
- In scope: pool health, connection liveness, SQL/pool logging hygiene, readiness checks, failure handling.
|
||||
- Out of scope: full APM stack design and vendor-specific monitoring platform setup.
|
||||
|
||||
---
|
||||
|
||||
## Rules
|
||||
|
||||
- Enable connection liveness strategy (`pool_pre_ping=True`) for long-running services.
|
||||
- Keep DB health checks out of liveness; include dependency checks in readiness.
|
||||
- Centralize engine options and logging configuration.
|
||||
- Avoid noisy SQL debug logging in production defaults.
|
||||
- Treat disconnect handling as a first-class test scenario.
|
||||
|
||||
---
|
||||
|
||||
## Recommended Baseline
|
||||
|
||||
```python
|
||||
engine = create_async_engine(
|
||||
settings.database_url,
|
||||
pool_pre_ping=True,
|
||||
# Tune only from measured behavior:
|
||||
# pool_size=10,
|
||||
# max_overflow=20,
|
||||
# pool_timeout=30,
|
||||
# pool_recycle=1800,
|
||||
)
|
||||
```
|
||||
|
||||
Operational guidance:
|
||||
|
||||
- `pool_pre_ping=True` for stale-connection resilience.
|
||||
- Introduce `pool_recycle` where backend/network idle timeout behavior warrants it.
|
||||
- Use structured app logs with request correlation and error context.
|
||||
|
||||
---
|
||||
|
||||
## Health Endpoint Policy
|
||||
|
||||
- `/healthz`: process is alive; no DB call required.
|
||||
- `/readyz`: application can currently serve traffic; include DB connectivity verification.
|
||||
|
||||
Readiness checks should be lightweight and bounded (timeouts), not heavy diagnostic queries.
|
||||
|
||||
---
|
||||
|
||||
## Failure Handling Guidance
|
||||
|
||||
- Handle transient disconnects with pool invalidation/reconnect semantics.
|
||||
- Keep one failed request from cascading into broad app instability.
|
||||
- Capture and log contextual DB errors with enough metadata for debugging.
|
||||
|
||||
---
|
||||
|
||||
## Anti-Patterns
|
||||
|
||||
- No readiness check for DB-dependent services.
|
||||
- Permanent debug SQL echo in production.
|
||||
- Per-handler ad hoc pool settings.
|
||||
- Assuming disconnect events are too rare to test.
|
||||
|
||||
---
|
||||
|
||||
## Operational Checks
|
||||
|
||||
- Engine creation is centralized and configured once.
|
||||
- Liveness/readiness behavior is documented and validated.
|
||||
- Pool settings are explicit, versioned, and reviewed.
|
||||
- DB-related errors produce actionable logs.
|
||||
|
||||
---
|
||||
|
||||
## Testing Checks
|
||||
|
||||
- Readiness endpoint test covers healthy and unhealthy DB states.
|
||||
- Integration test simulates disconnect/reconnect behavior.
|
||||
- Load/concurrency tests validate pool behavior under stress.
|
||||
@@ -0,0 +1,404 @@
|
||||
# Async SQLAlchemy Session Management
|
||||
|
||||
!!! info "Primary sources"
|
||||
- [Python `asynccontextmanager`](https://docs.python.org/3/library/contextlib.html#contextlib.asynccontextmanager)
|
||||
- [Python `functools.cache`](https://docs.python.org/3/library/functools.html#functools.cache)
|
||||
- [Python `inspect.signature`](https://docs.python.org/3/library/inspect.html#inspect.signature)
|
||||
- [SQLAlchemy asyncio extension](https://docs.sqlalchemy.org/en/21/orm/extensions/asyncio.html)
|
||||
- [SQLAlchemy session basics](https://docs.sqlalchemy.org/en/21/orm/session_basics.html)
|
||||
- [`nicegui-db` session implementation](https://forgejo.john-stream.com/john/nicegui-db/src/commit/126bc26ad8635a86bacf684d7bda409230347597/src/nicegui_db/db/session.py)
|
||||
|
||||
---
|
||||
|
||||
## Purpose
|
||||
|
||||
Define one canonical session model for SQLAlchemy asyncio:
|
||||
|
||||
- configure a lifespan-owned factory or resolve a URL-keyed cached factory,
|
||||
- create one AsyncSession per task or unit of work,
|
||||
- let callers supply a session when they already own the scope,
|
||||
- never share one AsyncSession across concurrent tasks.
|
||||
|
||||
---
|
||||
|
||||
## Scope and Non-Goals
|
||||
|
||||
- In scope: session factory creation, task scoping, and transaction demarcation.
|
||||
- Out of scope: framework dependency wiring, ORM model design, query optimization strategy, and schema migration tooling.
|
||||
|
||||
---
|
||||
|
||||
## Rules
|
||||
|
||||
- Create the application `async_sessionmaker` inside `database_scope()` and store it in application state for request dependencies.
|
||||
- Use `get_session_factory(db_url)` and `resolve_session_factory()` for standalone decorated operations that do not receive the application factory.
|
||||
- Use a fresh AsyncSession for each task or explicit unit of work.
|
||||
- Let reusable service functions accept `AsyncSession | None` and apply `@with_session` when standalone invocation is useful.
|
||||
- Pass an `AsyncSession` directly when composing several calls under one caller-owned scope.
|
||||
- Borrow a caller-provided session without beginning, closing, committing, or rolling it back.
|
||||
- Do not share AsyncSession across `asyncio.gather()` or parallel tasks.
|
||||
- Prefer direct dependency injection over global scoped-session patterns in new code.
|
||||
- Use explicit transaction boundaries (`async with session.begin():`) for writes.
|
||||
- Use `db_transaction_scope()` when a standalone operation must own engine, factory, session, and transaction lifetimes together.
|
||||
- Use `begin_nested()` directly and only when partial rollback through a database SAVEPOINT is required.
|
||||
|
||||
---
|
||||
|
||||
## Sessions and Transactions
|
||||
|
||||
A session and a transaction solve related but different problems:
|
||||
|
||||
| Concept | Responsibility | Typical lifetime |
|
||||
| --- | --- | --- |
|
||||
| `AsyncSession` | Provides the ORM workspace: executes queries, tracks loaded and changed objects in its identity map, and flushes pending changes. It also coordinates access to a database connection. | One task or explicit unit of work. |
|
||||
| Transaction | Defines the atomic database boundary: all work inside it commits together on success or rolls back together on failure. | One complete operation that must have a single outcome. |
|
||||
|
||||
A transaction belongs to a session; it is not an alternative to one. The session is the interface used by application and data-access code, while the transaction determines when that work becomes permanent. A session may coordinate sequential transactions during its lifetime, although short-lived application scopes commonly use one session for one transaction.
|
||||
|
||||
Use a session without a helper-owned commit boundary for independent reads or lower-level functions that must participate in whatever transaction their caller controls:
|
||||
|
||||
```python
|
||||
async with session_factory() as session:
|
||||
item = await find_item(session, item_id)
|
||||
```
|
||||
|
||||
Use an explicit transaction for writes, read-modify-write operations, or several statements that must succeed or fail as one unit:
|
||||
|
||||
```python
|
||||
async with session_factory.begin() as session:
|
||||
order = await create_order(session, order_data)
|
||||
await reserve_inventory(session, order)
|
||||
```
|
||||
|
||||
SQLAlchemy sessions use [autobegin](https://docs.sqlalchemy.org/en/21/orm/session_basics.html#auto-begin), so the first database operation normally starts a transaction even for a read. Therefore, “session-only” means that the surrounding helper owns only session lifetime and does not promise to commit; it does not mean that no database transaction exists. Closing such a session releases its resources and rolls back any unfinished transaction. An explicit `begin()` is valuable when application code must make the atomic boundary and commit ownership visible.
|
||||
|
||||
For most read-only operations, a session context is sufficient. Use an explicit transaction for reads when they need a defined consistency boundary, participate in a larger atomic operation, or use locking such as `SELECT ... FOR UPDATE`.
|
||||
|
||||
---
|
||||
|
||||
## Session Factory Mechanics
|
||||
|
||||
An `async_sessionmaker[AsyncSession]` is a reusable configuration object and callable session producer. It stores how sessions should be created, including the engine binding and options such as `expire_on_commit=False`. It is not itself a session, connection, or transaction, and calling it does not make a shared global `AsyncSession`.
|
||||
|
||||
The template exposes two construction paths with the same session options.
|
||||
|
||||
The application-owned path creates a factory inside the engine lifecycle:
|
||||
|
||||
```python
|
||||
from collections.abc import AsyncGenerator
|
||||
from contextlib import asynccontextmanager
|
||||
|
||||
from sqlalchemy.ext.asyncio import async_sessionmaker
|
||||
from sqlmodel.ext.asyncio.session import AsyncSession
|
||||
|
||||
from .engine import engine_scope
|
||||
|
||||
type SessionFactory = async_sessionmaker[AsyncSession]
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def database_scope(
|
||||
db_url: str,
|
||||
*,
|
||||
auto_flush: bool = True,
|
||||
) -> AsyncGenerator[SessionFactory]:
|
||||
async with engine_scope(db_url) as engine:
|
||||
yield async_sessionmaker(
|
||||
bind=engine,
|
||||
class_=AsyncSession,
|
||||
expire_on_commit=False,
|
||||
autoflush=auto_flush,
|
||||
)
|
||||
```
|
||||
|
||||
FastAPI lifespan enters this path once and stores the yielded factory on application state. The factory must not outlive the scope because its bound engine is disposed on exit.
|
||||
|
||||
The standalone path caches a factory by URL and `auto_flush` policy:
|
||||
|
||||
```python
|
||||
from functools import cache
|
||||
|
||||
from .engine import get_engine
|
||||
|
||||
|
||||
@cache
|
||||
def get_session_factory(
|
||||
db_url: str,
|
||||
*,
|
||||
auto_flush: bool = True,
|
||||
) -> SessionFactory:
|
||||
return async_sessionmaker(
|
||||
bind=get_engine(db_url),
|
||||
class_=AsyncSession,
|
||||
expire_on_commit=False,
|
||||
autoflush=auto_flush,
|
||||
)
|
||||
|
||||
|
||||
def resolve_session_factory(settings: Settings | None = None) -> SessionFactory:
|
||||
settings = settings or get_settings()
|
||||
db_url = get_database_url(settings)
|
||||
return get_session_factory(db_url)
|
||||
```
|
||||
|
||||
This path lets framework-independent helpers resolve one stable factory without receiving it through every call. The tradeoff is hidden configuration resolution and a second lifecycle mechanism. `dispose_engine()` clears `get_engine`'s cache but does not clear `get_session_factory`'s cache in the template. A cached factory remains bound to the disposed engine object; SQLAlchemy can create a new pool when that engine is used again, but a later `database_scope()` for the same URL can own a different engine. Treat cached standalone resolution as process-lifetime convenience, avoid repeated application lifecycles in one process, and clear both caches together if the template evolves to support them.
|
||||
|
||||
Each call to `session_factory()` creates a distinct `AsyncSession`. The caller that invokes the factory owns that session lifetime and must close it, normally with `async with`:
|
||||
|
||||
```python
|
||||
async with session_factory() as session:
|
||||
...
|
||||
```
|
||||
|
||||
The factory can be shared across operations and tasks. Sessions produced by it cannot be shared across concurrent tasks.
|
||||
|
||||
Passing the application factory directly has three useful consequences:
|
||||
|
||||
- Lower layers do not resolve settings or global resources.
|
||||
- Tests can inject a test factory directly through `session_scope(session_factory=...)` or FastAPI state.
|
||||
- Transaction ownership remains independent of engine construction.
|
||||
|
||||
---
|
||||
|
||||
## Database and Convenience Scopes
|
||||
|
||||
The template provides three framework-independent context managers:
|
||||
|
||||
```python
|
||||
from collections.abc import AsyncGenerator
|
||||
from contextlib import asynccontextmanager
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def db_session_scope(
|
||||
db_url: str | None = None,
|
||||
) -> AsyncGenerator[AsyncSession]:
|
||||
db_url = db_url or resolve_database_url()
|
||||
async with database_scope(db_url) as session_factory, session_factory() as session:
|
||||
yield session
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def db_transaction_scope(
|
||||
db_url: str | None = None,
|
||||
) -> AsyncGenerator[AsyncSession]:
|
||||
db_url = db_url or resolve_database_url()
|
||||
async with database_scope(db_url) as session_factory, session_factory.begin() as session:
|
||||
yield session
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def session_scope(
|
||||
*,
|
||||
settings: Settings | None = None,
|
||||
session_factory: SessionFactory | None = None,
|
||||
session: AsyncSession | None = None,
|
||||
) -> AsyncGenerator[AsyncSession]:
|
||||
if session is not None:
|
||||
yield session
|
||||
return
|
||||
|
||||
session_factory = session_factory or resolve_session_factory(settings=settings)
|
||||
async with session_factory() as owned_session:
|
||||
yield owned_session
|
||||
```
|
||||
|
||||
`db_session_scope()` owns a complete temporary database lifecycle and a session but does not commit. `db_transaction_scope()` owns the same resources plus a root transaction that commits on successful exit and rolls back on exception. Both initialize the schema by default because `database_scope()` enters `engine_scope()` with its default `initialize=True`. They are appropriate for scripts, commands, and isolated operations, not per-request use inside an already-running application.
|
||||
|
||||
`session_scope()` is the borrow-or-create helper. Its precedence is supplied session, supplied factory, then settings-based cached factory resolution. A supplied session remains entirely caller-owned; the helper does not require an active transaction and does not begin, commit, roll back, or close it. An owned session is closed on exit, and unfinished autobegun work rolls back.
|
||||
|
||||
Passing `session=None` is the same as omitting the session for `session_scope()` and therefore creates a session. This differs from `with_session`, which tests whether the argument name was bound rather than whether its value is non-null.
|
||||
|
||||
## Signature-Aware Session Injection
|
||||
|
||||
`with_session` allows one async function to support standalone calls and explicit composition:
|
||||
|
||||
```python
|
||||
from collections.abc import Awaitable
|
||||
from collections.abc import Callable
|
||||
from functools import wraps
|
||||
from inspect import signature
|
||||
|
||||
|
||||
def with_session[**P, R](
|
||||
func: Callable[P, Awaitable[R]],
|
||||
) -> Callable[P, Awaitable[R]]:
|
||||
sig = signature(func)
|
||||
|
||||
@wraps(func)
|
||||
async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
|
||||
bound = sig.bind_partial(*args, **kwargs)
|
||||
|
||||
if "session" in bound.arguments:
|
||||
return await func(*args, **kwargs)
|
||||
|
||||
async with resolve_session_factory()() as session:
|
||||
bound.arguments["session"] = session
|
||||
return await func(*bound.args, **bound.kwargs)
|
||||
|
||||
return wrapper
|
||||
```
|
||||
|
||||
The function must be async and expose a parameter named exactly `session`. The decorator preserves metadata with `wraps()`, binds positional and keyword arguments through the original signature, and injects a fresh session only when the caller omitted that argument.
|
||||
|
||||
The distinction between omitted and explicit `None` is deliberate in the implementation:
|
||||
|
||||
- `await operation()` injects and owns a session.
|
||||
- `await operation(session=existing_session)` borrows the caller's session.
|
||||
- `await operation(None)` or `await operation(session=None)` forwards `None` without injection.
|
||||
|
||||
The decorated function therefore types the parameter as `AsyncSession | None = None` but should assert or guard after decoration. Explicit `None` is not a request for injection. This preserves ordinary Python call binding, but it means wrappers or callers must omit the argument instead of forwarding a nullable value.
|
||||
|
||||
`with_session` owns session lifetime only. It does not begin or commit a transaction, so it is naturally suited to reads. Decorated writes must either manage a visible transaction or be called with a session from `db_transaction_scope()` or another caller-owned transaction. Prefer explicit factory or session injection when lifecycle transparency and test substitution matter more than call-site convenience.
|
||||
|
||||
---
|
||||
|
||||
## Function and Service Boundaries
|
||||
|
||||
Template service functions support both standalone and composed use by combining `@with_session` with an optional parameter:
|
||||
|
||||
```python
|
||||
from sqlmodel import func
|
||||
from sqlmodel import select
|
||||
|
||||
|
||||
@with_session
|
||||
async def count_items(session: AsyncSession | None = None) -> int:
|
||||
assert session is not None, "Session must be provided by with_session decorator"
|
||||
result = await session.exec(select(func.count()).select_from(Item))
|
||||
return result.one()
|
||||
```
|
||||
|
||||
The standalone call injects and closes a session:
|
||||
|
||||
```python
|
||||
count = await count_items()
|
||||
```
|
||||
|
||||
A larger use case passes one caller-owned session through several decorated functions:
|
||||
|
||||
```python
|
||||
async with session_factory.begin() as session:
|
||||
count = await count_items(session)
|
||||
await create_item(payload, session=session)
|
||||
```
|
||||
|
||||
The decorator sees the bound `session` argument and leaves all ownership with the caller. It never creates a SAVEPOINT or nested transaction.
|
||||
|
||||
For low-level helpers that should never resolve settings, require a non-optional session and leave them undecorated. Application service objects may store the immutable session factory, but they must not store a mutable session:
|
||||
|
||||
```python
|
||||
class ItemService:
|
||||
def __init__(self, session_factory: SessionFactory) -> None:
|
||||
self.session_factory = session_factory
|
||||
|
||||
async def find(self, item_id: int) -> Item | None:
|
||||
async with self.session_factory() as session:
|
||||
return await find_item(session, item_id)
|
||||
```
|
||||
|
||||
Code that already owns a transaction should call the session-required function directly. Repositories should normally remain in that session-required layer; the service or use-case boundary owns standalone session creation. This avoids optional-session APIs spreading into every data-access function.
|
||||
|
||||
---
|
||||
|
||||
## SAVEPOINTs and Partial Failure
|
||||
|
||||
Use [`begin_nested()`](https://docs.sqlalchemy.org/en/21/orm/session_transaction.html#using-savepoint) only when failure inside one portion of an operation should roll back that portion while preserving the outer transaction:
|
||||
|
||||
```python
|
||||
async with db_transaction_scope() as session:
|
||||
order = await insert_order(session, payload)
|
||||
|
||||
try:
|
||||
async with session.begin_nested():
|
||||
await apply_optional_discount(session, order)
|
||||
except DiscountError:
|
||||
pass
|
||||
|
||||
await reserve_inventory(session, order)
|
||||
```
|
||||
|
||||
Important SAVEPOINT semantics:
|
||||
|
||||
- `begin_nested()` starts a root transaction if one is not already active, so call it inside a visible outer transaction when that ownership matters.
|
||||
- Entering `begin_nested()` unconditionally flushes pending session state, regardless of the `autoflush` setting.
|
||||
- Successful exit releases the SAVEPOINT; it does not commit the outer transaction.
|
||||
- Exceptional exit rolls back to the SAVEPOINT and leaves the outer transaction active.
|
||||
- In SQLAlchemy 2.x, `session.commit()` commits the outermost transaction. Never call it to release a SAVEPOINT; let the nested context manager manage its transaction handle.
|
||||
|
||||
Do not create a SAVEPOINT merely because one service calls another. SAVEPOINTs add database work and alter flush and error-recovery behavior. Use them only for explicit partial-failure requirements such as skipping one conflicting row while retaining the rest of a batch.
|
||||
|
||||
---
|
||||
|
||||
## Framework Integration
|
||||
|
||||
Keep framework adapters outside these session primitives. See [FastAPI database integration](fastapi.md) for lifespan ownership, `Annotated` dependency aliases, and read-versus-write request sessions.
|
||||
|
||||
---
|
||||
|
||||
## Configuration Guidance
|
||||
|
||||
- `expire_on_commit=False` is commonly preferred in asyncio applications to reduce accidental post-commit reload behavior.
|
||||
- `AsyncSession.refresh()` is preferred over broad expiration patterns when state refresh is needed.
|
||||
- [`async_sessionmaker.begin()`](https://docs.sqlalchemy.org/en/21/orm/extensions/asyncio.html#sqlalchemy.ext.asyncio.async_sessionmaker.begin) is a concise option when one scope must create a session, begin a transaction, commit on success, roll back on failure, and close. Do not use it when borrowing a caller's session.
|
||||
|
||||
## SQLModel Alignment
|
||||
|
||||
- Use SQLModel as the default model and statement layer while keeping the same session ownership model: one `async_sessionmaker`, one `AsyncSession` per task or unit of work.
|
||||
- SQLModel does not replace SQLAlchemy async lifecycle primitives; it provides model declaration, validation, and typing ergonomics on top of them.
|
||||
- Do not mix ad hoc session construction with the canonical session factory.
|
||||
|
||||
---
|
||||
|
||||
## Concurrency Rules
|
||||
|
||||
- One session per concurrent task.
|
||||
- If work fans out into parallel tasks, each task receives its own AsyncSession.
|
||||
- Pass sessions explicitly to service functions; avoid mutable global session state.
|
||||
|
||||
---
|
||||
|
||||
## Anti-Patterns
|
||||
|
||||
- A singleton/global AsyncSession reused across tasks or operations.
|
||||
- Sharing one AsyncSession across parallel tasks.
|
||||
- Passing an application-global AsyncSession to a repository constructor.
|
||||
- Creating a new `async_sessionmaker` in each operation.
|
||||
- Retaining a session factory after its bound engine scope exits.
|
||||
- Using cached standalone factory resolution when the application factory is already available.
|
||||
- Assuming `with_session` starts or commits a transaction.
|
||||
- Forwarding `session=None` to a decorated function when injection was intended.
|
||||
- Closing or committing a session supplied by the caller.
|
||||
- Silently starting or committing a transaction on a supplied session.
|
||||
- Creating a SAVEPOINT for ordinary nested service calls.
|
||||
- Hiding root transaction, joined transaction, and SAVEPOINT behavior behind one mode-driven `atomic_scope()` helper.
|
||||
- Calling `session.commit()` inside a SAVEPOINT scope.
|
||||
- Mixing commit/rollback ownership across layers without a declared boundary.
|
||||
|
||||
---
|
||||
|
||||
## Operational Checks
|
||||
|
||||
- The FastAPI application factory is created inside `database_scope()` and does not outlive its bound engine.
|
||||
- Cached standalone factories are used only where application-state injection is unavailable.
|
||||
- `session_scope()` precedence is supplied session, supplied factory, then settings-based resolution.
|
||||
- Decorated functions receive injection only when the `session` argument is omitted.
|
||||
- No code path creates AsyncSession in module import side effects.
|
||||
- Concurrent jobs and operations each create task-local sessions.
|
||||
|
||||
---
|
||||
|
||||
## Testing Checks
|
||||
|
||||
- Service constructors accept a test session factory without framework startup.
|
||||
- Session-taking access functions accept a transaction-scoped test session directly.
|
||||
- `session_scope()` tests cover supplied-session, supplied-factory, and settings-resolution precedence.
|
||||
- `db_transaction_scope()` tests verify commit on success, rollback on failure, session closure, engine disposal, and cache cleanup.
|
||||
- `with_session` tests cover omitted, positional, keyword, and explicit-`None` session arguments.
|
||||
- Composition tests verify decorated service calls borrow one caller-owned session without committing it.
|
||||
- SAVEPOINT tests verify local rollback preserves the outer transaction and successful exit does not commit it.
|
||||
- Tests that depend on SAVEPOINT timing account for `begin_nested()` flushing pending state on entry.
|
||||
- Rollback behavior is verified for failed write units.
|
||||
- Parallel-task tests verify no shared AsyncSession instances.
|
||||
- Lifecycle tests confirm schema initialization, factory availability, deterministic teardown, and expected cache behavior.
|
||||
|
||||
@@ -0,0 +1,124 @@
|
||||
# SQLModel-First Modeling and Async Boundaries
|
||||
|
||||
!!! info "Primary sources"
|
||||
- [SQLModel documentation](https://sqlmodel.tiangolo.com/)
|
||||
- [SQLModel features](https://sqlmodel.tiangolo.com/features/)
|
||||
- [SQLModel advanced guide](https://sqlmodel.tiangolo.com/advanced/)
|
||||
- [SQLModel FastAPI session dependency tutorial](https://sqlmodel.tiangolo.com/tutorial/fastapi/session-with-dependency/)
|
||||
- [SQLModel release notes](https://sqlmodel.tiangolo.com/release-notes/)
|
||||
- [SQLAlchemy asyncio extension](https://docs.sqlalchemy.org/en/21/orm/extensions/asyncio.html)
|
||||
|
||||
??? abstract "Decision metadata"
|
||||
- Status: adopted
|
||||
- Decision level: mandatory
|
||||
- Applies to: api-runtime, workers, tests
|
||||
- Last reviewed: 2026-08-06
|
||||
|
||||
---
|
||||
|
||||
## Purpose
|
||||
|
||||
Define SQLModel as the primary model layer for async FastAPI applications and explain how it composes with SQLAlchemy's async runtime.
|
||||
|
||||
SQLModel is designed for FastAPI, built on Pydantic and SQLAlchemy, and intended to minimize duplication while preserving the capabilities of both. Async engine, session, transaction, and loading behavior still follow SQLAlchemy's asyncio contract.
|
||||
|
||||
---
|
||||
|
||||
## Scope and Non-Goals
|
||||
|
||||
- In scope: table models, API data models, SQLAlchemy interoperability, async session usage, and exception criteria.
|
||||
- Out of scope: replacing SQLAlchemy's async runtime primitives or claiming that synchronous tutorial examples are async patterns.
|
||||
|
||||
---
|
||||
|
||||
## Rules
|
||||
|
||||
- Default to SQLModel for new table models and API data models.
|
||||
- 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.
|
||||
- Use SQLAlchemy relationship loading options explicitly on async paths.
|
||||
|
||||
---
|
||||
|
||||
## Recommended Patterns
|
||||
|
||||
### Pattern A: Data model split for API boundaries
|
||||
|
||||
Use distinct models for persistence and external contracts.
|
||||
|
||||
```python
|
||||
from sqlmodel import Field, SQLModel
|
||||
|
||||
|
||||
class UserBase(SQLModel):
|
||||
email: str
|
||||
display_name: str
|
||||
|
||||
|
||||
class User(UserBase, table=True):
|
||||
id: int | None = Field(default=None, primary_key=True)
|
||||
|
||||
|
||||
class UserCreate(UserBase):
|
||||
pass
|
||||
|
||||
|
||||
class UserRead(UserBase):
|
||||
id: int
|
||||
```
|
||||
|
||||
### Pattern B: Keep SQLModel models with the async runtime
|
||||
|
||||
```python
|
||||
from sqlmodel import select
|
||||
|
||||
async with database_scope(settings.database_url) as session_factory:
|
||||
async with session_factory() as session:
|
||||
users = (await session.exec(select(User))).all()
|
||||
```
|
||||
|
||||
`database_scope()` enters the cached engine lifecycle, initializes registered SQLModel metadata by default, and yields the application session factory while SQLModel supplies the model and statement layer. `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.
|
||||
|
||||
---
|
||||
|
||||
## Interoperability Notes
|
||||
|
||||
- A SQLModel table model is a SQLAlchemy model and can participate in SQLAlchemy relationships, statements, loader options, and sessions.
|
||||
- A SQLModel model is also a Pydantic model; non-table models are useful for request and response contracts.
|
||||
- SQLModel's official FastAPI dependency tutorial currently uses synchronous `Session`; translate the ownership pattern, not the concrete session type, for async applications.
|
||||
- SQLModel's advanced guide still lists dedicated async documentation as future work, so use SQLAlchemy's asyncio documentation as the authority for runtime mechanics.
|
||||
- Prefer one query style per module to reduce cognitive overhead.
|
||||
- Keep loader strategies explicit in async paths to avoid implicit I/O surprises.
|
||||
|
||||
---
|
||||
|
||||
## Anti-Patterns
|
||||
|
||||
- Treating SQLModel as an alternative to SQLAlchemy rather than a layer built on it.
|
||||
- Copying a synchronous `Session` example into an async request path.
|
||||
- Constructing sessions in handlers instead of using the application session factory.
|
||||
- Mixing multiple query/session idioms within the same module without clear conventions.
|
||||
|
||||
---
|
||||
|
||||
## Operational Checks
|
||||
|
||||
- New model modules are SQLModel-first; exceptions state the unsupported need or constraint.
|
||||
- Session/transaction ownership remains consistent across both model styles.
|
||||
- Table, create, update, and public models share fields intentionally without exposing persistence-only data.
|
||||
|
||||
---
|
||||
|
||||
## Testing Checks
|
||||
|
||||
- Module-level tests verify CRUD semantics for SQLModel models through `AsyncSession`.
|
||||
- API tests verify response/request model behavior for SQLModel-based endpoints.
|
||||
- Relationship tests verify async loader strategies do not depend on implicit I/O.
|
||||
|
||||
---
|
||||
|
||||
## Version Checks
|
||||
|
||||
- Verify installed SQLModel, SQLAlchemy, and Pydantic versions together when using newly added typing or ORM features.
|
||||
@@ -0,0 +1,59 @@
|
||||
# <Concept Title>
|
||||
|
||||
!!! info "Primary sources"
|
||||
- Primary source: `<primary source URL>`
|
||||
- Secondary source: `<secondary source URL>`
|
||||
|
||||
??? abstract "Decision metadata"
|
||||
- Status: draft|adopted|deprecated
|
||||
- Decision level: advisory|mandatory
|
||||
- Applies to: api-runtime|workers|tests
|
||||
- Last reviewed: YYYY-MM-DD
|
||||
|
||||
---
|
||||
|
||||
## Purpose
|
||||
|
||||
Describe what this concept governs and why it exists.
|
||||
|
||||
## Scope and Non-Goals
|
||||
|
||||
- In scope:
|
||||
- Out of scope:
|
||||
|
||||
---
|
||||
|
||||
## Rules
|
||||
|
||||
- Rule 1
|
||||
- Rule 2
|
||||
|
||||
---
|
||||
|
||||
## Recommended Pattern
|
||||
|
||||
```python
|
||||
# minimal example
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Anti-Patterns
|
||||
|
||||
- Anti-pattern 1
|
||||
- Anti-pattern 2
|
||||
|
||||
---
|
||||
|
||||
## Operational Checks
|
||||
|
||||
- Check 1
|
||||
- Check 2
|
||||
|
||||
---
|
||||
|
||||
## Testing Checks
|
||||
|
||||
- Test 1
|
||||
- Test 2
|
||||
|
||||
@@ -0,0 +1,172 @@
|
||||
# Testing Database Targets and Data
|
||||
|
||||
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 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.
|
||||
|
||||
## Shared Construction Primitives
|
||||
|
||||
Make the application factory accept a database URL or settings object. Production, workers, and ordinary integration tests enter the same [`database_scope()`](session.md#database-and-convenience-scopes). Tests enter the lower-level [`engine_scope()`](engine.md#owning-engine-scope) only when they need direct engine or connection ownership for schema setup, an outer transaction, or engine-specific assertions:
|
||||
|
||||
```python
|
||||
from collections.abc import AsyncGenerator
|
||||
|
||||
import pytest_asyncio
|
||||
from sqlalchemy.ext.asyncio import AsyncEngine
|
||||
|
||||
from .engine import engine_scope
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(scope="session", loop_scope="session")
|
||||
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 `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 schema initialization, deterministic disposal, and engine-cache cleanup 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 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 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(scope="function", loop_scope="session")
|
||||
async def session_factory(test_engine: AsyncEngine) -> AsyncGenerator[SessionFactory]:
|
||||
async with test_engine.connect() as connection:
|
||||
transaction = await connection.begin()
|
||||
factory = async_sessionmaker(
|
||||
bind=connection,
|
||||
class_=AsyncSession,
|
||||
expire_on_commit=False,
|
||||
join_transaction_mode="create_savepoint",
|
||||
)
|
||||
try:
|
||||
yield factory
|
||||
finally:
|
||||
await transaction.rollback()
|
||||
```
|
||||
|
||||
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 pass a caller-owned session into decorated or undecorated service functions, derive that session from the same factory:
|
||||
|
||||
```python
|
||||
@pytest_asyncio.fixture(scope="function", loop_scope="session")
|
||||
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()` gives test code one visible transaction from the start. 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` so the production `SessionDep` retains its session-creation and cleanup 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 .fastapi import _get_session_factory
|
||||
from .session import SessionFactory
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
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_factory] = get_test_session_factory
|
||||
try:
|
||||
yield app
|
||||
finally:
|
||||
app.dependency_overrides.pop(_get_session_factory, None)
|
||||
```
|
||||
|
||||
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
|
||||
|
||||
### 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.
|
||||
|
||||
`engine_scope()` imports the model package and creates the schema by default, then disposes the engine and clears cached resolution deterministically:
|
||||
|
||||
```python
|
||||
from collections.abc import AsyncGenerator
|
||||
|
||||
import pytest_asyncio
|
||||
from sqlalchemy.ext.asyncio import AsyncEngine
|
||||
from .engine import engine_scope
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(scope="session", loop_scope="session")
|
||||
async def test_engine() -> AsyncGenerator[AsyncEngine]:
|
||||
async with engine_scope("sqlite+aiosqlite://") as engine:
|
||||
yield engine
|
||||
```
|
||||
|
||||
### 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 `database_scope()` unless a test explicitly needs lower-level engine or connection ownership.
|
||||
- Every test or fixture scope owns its override, session factory, connection, transaction, and session cleanup; the session-scoped engine fixture owns disposal and cache cleanup.
|
||||
- Request tests override `_get_session_factory`, preserving production request-session creation and cleanup 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.
|
||||
|
||||
## 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/)
|
||||
- [pytest-asyncio fixtures](https://pytest-asyncio.readthedocs.io/en/stable/how-to-guides/index.html)
|
||||
@@ -0,0 +1,105 @@
|
||||
# Async Transaction Boundaries
|
||||
|
||||
!!! info "Primary sources"
|
||||
- [SQLAlchemy transactions](https://docs.sqlalchemy.org/en/21/orm/session_transaction.html)
|
||||
- [SQLAlchemy asyncio extension](https://docs.sqlalchemy.org/en/21/orm/extensions/asyncio.html)
|
||||
- [SQLAlchemy connections](https://docs.sqlalchemy.org/en/21/core/connections.html)
|
||||
|
||||
??? abstract "Decision metadata"
|
||||
- Status: adopted
|
||||
- Decision level: mandatory
|
||||
- Applies to: api-runtime, workers, tests
|
||||
- Last reviewed: 2026-06-17
|
||||
|
||||
---
|
||||
|
||||
## Purpose
|
||||
|
||||
Define consistent transaction demarcation for async SQLAlchemy so write behavior is predictable, rollback semantics are clear, and concurrent request flows remain safe.
|
||||
|
||||
---
|
||||
|
||||
## Scope and Non-Goals
|
||||
|
||||
- In scope: transaction ownership, write/read policy, exception and rollback behavior, nested transaction guidance.
|
||||
- Out of scope: business-domain validation rules and cross-service distributed transactions.
|
||||
|
||||
---
|
||||
|
||||
## Rules
|
||||
|
||||
- Every mutating use case must run inside an explicit transaction boundary.
|
||||
- Prefer `async with session.begin():` for write units.
|
||||
- Keep transaction ownership at a service, use-case, or explicitly documented complete-operation boundary, not deep in helper internals.
|
||||
- An optional-session write may own one transaction when omitting the session clearly means standalone execution; a supplied session must remain caller-owned.
|
||||
- Read paths should not auto-upgrade into hidden write behavior.
|
||||
- On exception in a transaction block, rely on rollback semantics and propagate or map exceptions intentionally.
|
||||
|
||||
---
|
||||
|
||||
## Recommended Patterns
|
||||
|
||||
### Pattern A: Single write unit
|
||||
|
||||
```python
|
||||
async def create_order(session: AsyncSession, payload: OrderIn) -> Order:
|
||||
async with session.begin():
|
||||
order = Order(...)
|
||||
session.add(order)
|
||||
# additional writes...
|
||||
return order
|
||||
```
|
||||
|
||||
### Pattern B: Explicit read flow
|
||||
|
||||
```python
|
||||
async def get_order(session: AsyncSession, order_id: UUID) -> Order | None:
|
||||
stmt = select(Order).where(Order.id == order_id)
|
||||
return await session.scalar(stmt)
|
||||
```
|
||||
|
||||
### Pattern C: Nested transaction (only when required)
|
||||
|
||||
```python
|
||||
async with session.begin():
|
||||
# outer transaction
|
||||
async with session.begin_nested():
|
||||
# savepoint-scoped operation
|
||||
...
|
||||
```
|
||||
|
||||
Use nested transactions only when partial failure semantics are explicitly required.
|
||||
|
||||
---
|
||||
|
||||
## Exception and Rollback Policy
|
||||
|
||||
- Write block fails: transaction context rolls back.
|
||||
- Caller decides whether to translate exception (for example to domain/API errors).
|
||||
- Do not swallow DB exceptions silently; map or re-raise intentionally.
|
||||
|
||||
---
|
||||
|
||||
## Anti-Patterns
|
||||
|
||||
- Multiple commits scattered across one logical use case.
|
||||
- Helper functions that commit or roll back without an explicit ownership contract.
|
||||
- Mixing implicit and explicit transaction styles in confusing ways.
|
||||
- Using savepoints as a default pattern rather than a targeted tool.
|
||||
|
||||
---
|
||||
|
||||
## Operational Checks
|
||||
|
||||
- All mutating services and complete operations declare one clear transaction boundary.
|
||||
- No repository or helper performs hidden direct commit calls; standalone ownership is expressed through a documented transaction scope.
|
||||
- Transaction style is consistent across handlers and workers.
|
||||
|
||||
---
|
||||
|
||||
## Testing Checks
|
||||
|
||||
- Success path test verifies expected durable writes.
|
||||
- Failure path test verifies rollback behavior.
|
||||
- Tests cover concurrency-sensitive write flows.
|
||||
- Savepoint usage (if present) has dedicated behavior tests.
|
||||
Reference in New Issue
Block a user