SQLAlchemy ORM skill for Claude Code

SQLAlchemy work is done well when models are fully typed and the database constraints are named and explicit, every Session has one obvious owner and a short transaction, every relationship a request touches is loaded on purpose (never by accident), and every Alembic migration was read before it ran. The ORM writes SQL for you; you still own that SQL.

First, identify the setup

python -c "import sqlalchemy, sys; print(sqlalchemy.__version__, sys.version)"
grep -rnE "session\.query\(|declarative_base\(|Column\(|backref=" --include=*.py . | head

Version facts to apply before writing code:

• 2.1 (Sep 2026, current): Python 3.11+. greenlet is no longer installed by default; async users must install sqlalchemy[asyncio]. A bare postgresql:// URL now uses psycopg 3 (was psycopg2): pin the driver explicitly with postgresql+psycopg:// or postgresql+psycopg2://. Autoflush now also happens before text() executions. select().filter_by() raises on names that exist in more than one FROM entity.
• 2.0: still maintained (2.0.5x). Everything below works on 2.0 unless noted.
• 1.x patterns (session.query(), declarative_base(), untyped Column) still run under 2.x as legacy. Match the codebase; when touching a module, migrate it to select() and Mapped[...] in a separate commit from behavior changes.

Models

from datetime import datetime
from sqlalchemy import ForeignKey, MetaData, String, func, DateTime
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    metadata = MetaData(naming_convention={
        "ix": "ix_%(column_0_label)s",
        "uq": "uq_%(table_name)s_%(column_0_name)s",
        "ck": "ck_%(table_name)s_%(constraint_name)s",
        "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
        "pk": "pk_%(table_name)s",
    })
    type_annotation_map = {datetime: DateTime(timezone=True)}

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(320), unique=True)
    name: Mapped[str | None]                                    # Optional -> NULL allowed
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    orders: Mapped[list["Order"]] = relationship(
        back_populates="user", lazy="raise", cascade="all, delete-orphan", passive_deletes=True)

class Order(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True)
    status: Mapped[str] = mapped_column(String(20), server_default="open")
    user: Mapped["User"] = relationship(back_populates="orders", lazy="raise")

Rules:

• Nullability comes from the annotation: Mapped[str] is NOT NULL, Mapped[str | None] is nullable.
• Set a naming convention on day one. Without it, Alembic cannot drop or alter unnamed constraints reliably and autogenerate misses them.
• server_default puts the default in the DDL (visible to raw SQL and other services); default= is Python-side only. Prefer server_default for timestamps and status.
• Store timestamps as DateTime(timezone=True); never naive datetimes on Postgres.
• Index foreign key columns you filter or join on (Postgres does not). Use back_populates on both sides, never the legacy backref=.
• lazy="raise" on relationships makes accidental lazy loads fail loudly in tests instead of silently becoming N+1 queries in production. Load explicitly per query.
• With ondelete="CASCADE" in the FK, add passive_deletes=True so the ORM lets the database delete children instead of loading them all first.
• For collections that can grow unbounded (events, audit rows), use WriteOnlyMapped[list["Event"]] with relationship(): it never loads implicitly and exposes .select(), .add(), .remove().

Engine and Session lifecycle

One engine per process, created at startup. One Session per unit of work (request, job, task), never shared across threads or asyncio tasks, closed at the end.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine(settings.DATABASE_URL, pool_pre_ping=True, pool_size=5, max_overflow=5)
SessionLocal = sessionmaker(engine, expire_on_commit=False)

Context	Pattern
FastAPI (sync)	Dependency with SessionLocal() as session: yield session; the handler or service calls session.commit() explicitly; leaving the block rolls back anything uncommitted
Flask	Flask-SQLAlchemy (scoped to the app context), or scoped_session(SessionLocal) plus Session.remove() in teardown_appcontext
Celery / RQ / cron job / script	with SessionLocal.begin() as session: inside the task body (commit on success, rollback on exception); chunk big loops
Async (FastAPI, aiohttp)	async_sessionmaker and async with per request (see Async below)

Hard rules:

• Never keep a module-level Session or one Session for a whole worker process. The identity map grows forever and holds stale objects.
• Do no network calls (HTTP, email, queues) while a transaction is open. Commit first, or use an outbox table.
• After os.fork() (gunicorn --preload, multiprocessing), call engine.dispose(close=False) in the child before first use so parent connections are not shared.
• expire_on_commit=False avoids reload queries (and DetachedInstanceError) when serializing after commit; attributes may be stale, so refresh when it matters.
• DetachedInstanceError means an unloaded attribute was touched after the Session closed. Fix by eager loading that attribute in the query, not by keeping the Session open longer.
• Keep (pool_size + max_overflow) x processes under the DB limit (connection-pooling).

Queries (2.0 style)

from sqlalchemy import select, func, update
from sqlalchemy.orm import selectinload, joinedload

user = session.get(User, user_id)                                   # PK lookup, uses identity map
users = session.scalars(select(User).where(User.email.ilike("%@acme.io"))).all()
row = session.execute(select(User.id, func.count(Order.id))
                      .join(Order).group_by(User.id)).all()        # tuples, not entities

stmt = (select(Order)
        .where(Order.status == "open")
        .options(joinedload(Order.user), selectinload(Order.items))
        .order_by(Order.id).limit(50))
orders = session.scalars(stmt).all()

total = session.scalar(select(func.count()).select_from(stmt.order_by(None).limit(None).subquery()))
locked = session.scalars(select(Account).where(Account.id == aid).with_for_update()).one()

• session.scalars() for entities, session.execute() for multi-column rows, session.scalar() for one value. .one() raises when zero or many; .first() hides duplicates, so use it only with an ORDER BY you mean.
• Results that joinedload a collection need .unique() before .all().
• Paginate with keyset (where(Order.id > last_id).order_by(Order.id).limit(n)) for deep pages; offset() scans and discards every skipped row.
• Never build SQL with f-strings: text("... WHERE id = :id") + params, or tstring() (2.1).

Loading strategy

Relationship in this query	Use
Many-to-one / one-to-one (order -> user)	joinedload(Order.user)
One-to-many / many-to-many (user -> orders)	selectinload(User.orders) (one extra IN query, no row multiplication)
Several collections at once	One selectinload per collection; never joinedload two collections (cartesian product)
You already JOIN to filter on the related table	contains_eager(Order.user) to reuse that join
Must not be touched in this code path	raiseload("*") as a final option
Nested	Chain: selectinload(User.orders).joinedload(Order.product)
Only a count or sum of children	Aggregate in SQL with func.count and group_by, not by loading children

To detect N+1 across a codebase (query logs, per-request counts), use n-plus-one-queries; this table is how to fix it in SQLAlchemy.

Async

from sqlalchemy.ext.asyncio import AsyncAttrs, async_sessionmaker, create_async_engine

engine = create_async_engine("postgresql+asyncpg://...", pool_pre_ping=True)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

class Base(AsyncAttrs, DeclarativeBase): ...

async def list_orders(user_id: int) -> list[Order]:
    async with SessionLocal() as session:
        result = await session.scalars(
            select(Order).where(Order.user_id == user_id).options(selectinload(Order.items)))
        return list(result)

• Lazy loading cannot run implicitly under asyncio; it raises MissingGreenlet. Eager load in the query, await obj.awaitable_attrs.items, or await session.refresh(obj, ["items"]).
• One AsyncSession per task. Never pass a session into asyncio.gather() branches.
• await session.run_sync(fn) runs legacy sync ORM code (with lazy loading) inside an async session.
• await engine.dispose() on shutdown (FastAPI lifespan), or you get “Event loop is closed” warnings.
• On 2.1, install sqlalchemy[asyncio] or greenlet explicitly.

Transactions

with SessionLocal.begin() as session:          # commits on success, rolls back on exception
    acct = session.get(Account, aid, with_for_update=True)
    acct.balance -= amount
    session.add(Ledger(account_id=aid, amount=-amount))

with SessionLocal() as session:
    for row in rows:
        try:
            with session.begin_nested():       # SAVEPOINT: one bad row does not abort the batch
                session.add(Import(**row))
        except IntegrityError:
            log.warning("duplicate %s", row["key"])
    session.commit()

• Set isolation per engine or per connection: engine.execution_options(isolation_level="SERIALIZABLE"). Retry serialization failures with backoff; the whole unit of work must be retried, not one statement.
• Prefer atomic SQL (update(Account).where(...).values(balance=Account.balance - x)) over read-modify-write. Lock design itself belongs to transactions-locking.

Bulk operations

Need	Pattern
Insert many rows	session.execute(insert(User), list_of_dicts) (keys are attribute names; batched via insertmanyvalues)
Insert and get rows back	session.scalars(insert(User).returning(User, sort_by_parameter_order=True), rows)
Update many rows by PK	session.execute(update(User), [{"id": 1, "name": "a"}, ...]) (every dict needs the full PK)
Same change for a filtered set	session.execute(update(User).where(User.active.is_(False)).values(archived=True))
Upsert on Postgres	postgresql.insert(User).values(rows).on_conflict_do_update(index_elements=[User.email], set_={...})

• Chunk large loads (for example 1,000-10,000 rows per execute, commit per chunk) and measure; one giant transaction holds locks and bloats WAL.
• session.add() in a loop of 100k objects builds 100k tracked objects. Use the bulk forms above for imports.
• WHERE-based update/delete default to synchronize_session="auto" (fetch via RETURNING where supported). Objects already loaded in the Session may otherwise be stale; use "fetch" or expire them.
• Bulk DML skips ORM events, default= callables that depend on objects, and relationship cascades. Use server_default and database FKs if those must apply.

Alembic

alembic revision --autogenerate -m "add order status"   # then READ the file
alembic upgrade head --sql > /tmp/review.sql            # offline SQL for review
alembic upgrade head
alembic check                                           # CI: fails if models and DB differ
alembic heads                                           # more than one -> alembic merge heads

env.py must set target_metadata = Base.metadata and import every model module. Consider compare_server_default=True in context.configure() (off by default; compare_type is on by default since 1.12).

Review every autogenerated file for what autogenerate does not detect or gets wrong:

• Column or table renames appear as drop + add (data loss). Rewrite as op.alter_column(..., new_column_name=...) / op.rename_table.
• Unnamed constraints, CHECK expression changes, sequences, and some server default changes are missing. Add them by hand.
• Postgres ENUM changes: adding a value needs op.execute("ALTER TYPE ... ADD VALUE ..."); autogenerate will not emit it.
• Data migrations: use op.execute() or lightweight sa.table() constructs, never import application models (they change later and break old migrations).
• downgrade() exists and is correct, or explicitly raises for irreversible steps.
• Large-table DDL (new NOT NULL column, index, type change) follows a lock-safe plan: hand that part to schema-migrations.

Never run alembic downgrade or alembic stamp against a shared or production database without explicit user confirmation; downgrades can drop columns and their data.

Testing

Test against the same database engine as production (Postgres in Docker or testcontainers), not SQLite. Wrap each test in an outer transaction that is rolled back:

import pytest
from sqlalchemy.orm import Session

@pytest.fixture(scope="session")
def engine():
    eng = create_engine(TEST_DATABASE_URL)
    Base.metadata.create_all(eng)          # or run `alembic upgrade head` once
    yield eng
    eng.dispose()

@pytest.fixture
def session(engine):
    with engine.connect() as conn:
        trans = conn.begin()
        sess = Session(bind=conn, join_transaction_mode="create_savepoint")
        try:
            yield sess                     # code under test may commit() freely
        finally:
            sess.close()
            trans.rollback()

Async: same shape with async with engine.connect(), AsyncSession(bind=conn, ...) and await conn.rollback(). Override the app’s session dependency to yield it.

Lock in query counts on hot paths:

from contextlib import contextmanager
from sqlalchemy import event

@contextmanager
def count_queries(engine):
    seen = []
    def on_exec(conn, cursor, statement, params, context, executemany):
        seen.append(statement)
    event.listen(engine, "before_cursor_execute", on_exec)   # async: engine.sync_engine
    try:
        yield seen
    finally:
        event.remove(engine, "before_cursor_execute", on_exec)

def test_order_list_is_constant_queries(client, engine):
    with count_queries(engine) as q:
        client.get("/orders?limit=50")
    assert len(q) <= 3, q

Gotchas

• session.query(...).get() is legacy; use session.get(Model, pk).
• Use .is_(None) in where(); Python x is None evaluates to a plain bool.
• in_([]) with an empty list emits an always-false predicate; guard optional filters so an empty list does not silently mean “match nothing” when you meant “no filter”.
• Mutating a JSON column in place (obj.data["k"] = 1) is not tracked unless the column uses MutableDict.as_mutable(JSONB); otherwise reassign the whole value.
• session.merge() issues a SELECT per object; it is not an upsert.

Verification

1. Type-check (mypy or pyright); Mapped[...] makes wrong attribute types visible.
2. Run tests with lazy="raise" in place: any accidental lazy load fails.
3. Assert query counts on changed endpoints; EXPLAIN anything slow (sql-query-tuning).
4. Migrations: alembic upgrade head on a copy of the production schema, alembic check clean, then downgrade and upgrade again in a scratch database.

Output

**Change:** Order listing moved to 2.0 select(); user via joinedload, items via selectinload
**Queries per request:** 51 -> 3 (asserted in tests/test_orders.py)
**Sessions:** request-scoped dependency; commit in service layer; no I/O inside transactions
**Migration:** 2026_10_20_add_order_status (reviewed: rename rewritten as alter_column, ENUM value added by hand)
**Verified:** pytest green on Postgres 17 container; alembic check clean; mypy clean