Internal: SQLAlchemy async and Alembic under FastAPI (repositories, sessions, migrations); read by work's executors before such code.
Resources
1Install
npx skillscat add idontlikedrinkpivo/web-dev-pipeline-codex-plugin/fastapi-sqlalchemy Install via the SkillsCat registry.
Codex runtime. Before using this skill, read
../executor-catalog/references/codex-runtime.md. It defines tool mappings, named-agent dispatch and stage-chat orchestration; use it for inherited Claude Code terms below.
FastAPI + SQLAlchemy async
Which layout governs — decide first
The cases (architecture document, Framework-first, existing repo) are intypescript/references/which-layout-governs.md, shared by the backend and test skills of both stacks — read it
first. SQLAlchemy's placement in them:
- An architecture document exists: the table below.
- Framework-first: the models are the storage, and the route or a small service queries them directly.
- Existing repo: a
src/<domain>/service.pystays where it is; a new rule imports neither FastAPI nor SQLAlchemy. - HTTP: routes, the problem+json error handlers, auth and settings are
fastapi; this skill ends where a repository returns an entity or raises a named error.
Layout with an architecture document
| What | Where | Rule |
|---|---|---|
| Entity, value object, named errors | <area>/domain/ |
Owns every invariant-table rule. No fastapi, sqlalchemy, pydantic import |
| Use case | <area>/application/ |
Only orchestrates: load through a port, call the entity method, save through a port. Holds no rule and never touches the session or a record |
| Port | application/ports.py (Modest) or application/ports/ (Full) |
typing.Protocol with the foundation §4 signatures |
| SQLAlchemy model | <area>/adapters/persistence/ (file name from foundation §5) |
A storage record. Never imported by domain/ or application/, never returned as the entity or as the HTTP response |
| Repository adapter + mapper | <area>/adapters/persistence/ |
Implements the port; to_entity(record) / to_record(entity) written by hand; returns entities |
| Read query | adapters/persistence/ |
Returns the read model its group's scenarios document (documentation/architecture/scenarios/<area>/<group>.md) names, not the entity |
| Router factory, request / response schemas | <area>/adapters/http/ |
See fastapi |
| Engine, sessionmaker, use-case construction, exception handlers | bootstrap/ |
May import everything; nothing imports it |
- No
service.pythat holds business logic beside an anemic entity: the rule belongs to the entity method its invariant-table row names. - Each mapper gets a round-trip test (
typescript/references/shared-test-rules.md→ 1, written bypytest-patterns). - Run the import-linter contracts from foundation §6 after the change (
uv run lint-imports, or the command the repo already uses).
# <area>/adapters/persistence/order_repository.py — shape only, names from foundation §4/§5
class SqlOrderRepository: # implements application.ports.OrderRepository
def __init__(self, session: AsyncSession) -> None:
self._session = session
async def get(self, order_id: OrderId) -> Order | None:
record = await self._session.scalar(
select(OrderRecord).options(selectinload(OrderRecord.lines)).where(OrderRecord.id == order_id)
)
return None if record is None else to_entity(record)save adds to_record(entity) for a new entity, or copies the entity's fields onto the record it loaded with the same options, then flushes or commits per the unit-of-work rule below.
Session and unit of work
- One
AsyncSessionper request (or per use-case run in a job). Never module-global, never shared between concurrent tasks. - Create the sessionmaker with
expire_on_commit=False, so returned objects stay readable after commit without implicit IO. - Commit explicitly at the unit-of-work boundary the design names: inside
repository.savefor a single-entity save, aUnitOfWorkadapter when a scenario saves several. Brownfield: where the repo already commits. - The commit happens before the response is sent. Code after
yieldin a dependency may run after the response, so a failing commit there still answers 2xx. If a dependency owns commit / rollback, make it finish before the response (references/version-checks.md→yield-dependency timing), roll back and re-raise on an exception, and let the code inside callflush()(database ids exist before serialization) instead ofcommit(). - The session reaches a route through the
DbSessionalias (fastapi→ Dependency aliases).
Models and column types
- Typed declarative (
Mapped[...],mapped_column). Column types, lengths, nullability,CHECK/UNIQUE/FKand indexes copy the schema document (documentation/db/schema.md, fromdb-schema-design) or, brownfield, the existing migrations. - Money is never
float/Float:Numeric(p, s)withDecimal, or integer minor units, whichever the schema says. Timestamps are timezone-aware. - The database is PostgreSQL as the schema document states;
DATABASE_URLhas no SQLite default. Tests run on PostgreSQL too (pytest-patterns).
Relationships under AsyncSession
- A lazy load performs IO on attribute access and raises
MissingGreenlet. Declare every relationshiplazy="raise"so a missing load fails in tests, not in production. - Load explicitly (
selectinload/joinedload) in every query whose result is mapped or serialized, including the object a create returns: re-select with options, or refresh the named attributes. A path that "works" because the related row is already in the session's identity map is the same bug waiting for another caller. - With an architecture document the repository loads everything its mapper reads, so no lazy attribute leaves the adapter.
Migrations
- The migrations are the SQL files in
documentation/db/migrations/(0001_init.sql,0002_…), written bydb-schema-design. Alembic only orders and records them: one revision per file, revision id = the file's number,down_revision= the previous one, andupgrade()executes that file's text read fromdocumentation/db/migrations/(path from the repo root, the same relative path inside the image). Never autogenerate DDL and never retype a file intoop.*calls — two copies of a migration drift. A committed SQL file is never edited; a change is the next file and the next revision. downgrade()undoes what the schema changelog row calls reversible; otherwise it raises with the file name.- A file with
CONCURRENTLY(its own file,db-schema-design/references/migration-safety.md) runs insidewith op.get_context().autocommit_block():— a revision runs in a transaction, and PostgreSQL refusesCREATE INDEX CONCURRENTLYinside one. The block commits what ran before it, soenv.pysetstransaction_per_migration=True: each revision is then its own transaction, and a failed one leaves the earlier ones recorded. - Never
metadata.create_all— not in application code and not in test fixtures: the tests build their schema with the same migrations (pytest-patterns), so a model that disagrees with the files fails there. The one exception is a brownfield repo with no migration files at all: its test fixture keepscreate_alluntil the first migration file exists, and the report says so. Migrations run before the app starts, not from the lifespan.version_table_schemais the project schema (db-schema-design§0). - After
alembic upgrade head,alembic check(or autogenerate into a scratch file) must find no difference between the models and the database: a difference is a model that does not matchschema.md. - No database up: render the SQL offline (
alembic upgrade head --sql) and read it.
Driver errors and install
- A constraint violation the repository can expect (a duplicate email, a dangling reference, an overlap the
database refuses) is classified by its SQLSTATE and constraint name read from the driver's error (e.orig.sqlstate;
where the constraint name sits is a version check), never from its message — and raised as a domain exception with fields
(raise BookingOverlap(room_id=…) from e); its HTTP status is the contract's for that named error (fastapi).
Never a bare re-raise of the driver's error as an answer, neverexcept Exception: pass(python). - A serialization failure (
40001) or a deadlock (40P01) propagates unchanged from the repository. A use case
whose transaction can lose to a concurrent one is wrapped inbootstrap/by one shared helper that reruns its
whole run — load → decide → save, with a fresh session — a bounded number of times (three, with a short random
wait); never one statement, never a loop in the repository, and the use case knows no driver codes. What is left
after the attempts the handler answers with the contract's 503 andRetry-After. Anything else the
repository does not classify propagates unchanged and ends as the contract's 500 (fastapi→ «Errors as
problem+json»). - Async SQLAlchemy needs
sqlalchemy[asyncio]: from 2.1 greenlet is no longer installed by default, and without it
the first async call fails with an error that names the missing greenlet library — install the extra.MissingGreenletis a different failure: IO started outside anawait, nearly always a lazy load — fix the
loading (above), not the install.
Version-sensitive points — references/version-checks.md. In a work packet their answers come inCURRENT LIB DOCS; where it says none, probe the installed version in the project instead of fetching docs.
Outside work, read the file and check each point you rely on against the installed version.
The reviewer checks
- The commit happens before the response is sent — not in code after a dependency's
yieldthat runs after
it; a failing commit must not answer 2xx. - Every relationship is
lazy="raise"and every query that is mapped or serialized loads what it reads
(selectinload/joinedload), the object a create returns included. - Migrations are the SQL files: one revision per file executing its text, no autogenerated or retyped DDL, no
edited committed file,CONCURRENTLYinsideautocommit_block(), nocreate_allanywhere — except the test
fixture of a brownfield repo with no migration files at all. domain/andapplication/import no SQLAlchemy; the repository returns entities through a mapper that has
a round-trip test.- An expected constraint violation becomes a named exception with fields,
from e, classified by SQLSTATE
and constraint name; nothing else is swallowed or turned into a client answer here. - One
AsyncSessionper request or job, never module-global, never shared by concurrent tasks.