/sqlalchemy-patterns

name: sqlalchemy-patterns
description: >-
  Comprehensive SQLAlchemy 2.0+ async + PostgreSQL patterns — declarative
  models, relationships, type-safe Mapped[ columns, querying, eager loading,
  transactions, and Alembic migrations. Use when working with SQLAlchemy; to
  write a model, declarative model, or ORM model; add a column, mapped_column,
  or Mapped[ annotation; define a relationship, foreign key, or index on a
  table; choose selectinload, joinedload, or load_only; configure an async
  session or call session.execute; fix N+1 queries; do a bulk insert or upsert;
  use JSONB or a PostgreSQL type; count rows; write a migration model; or write
  any SQLAlchemy query. Not for SQLAlchemy 1.x, non-async (sync) codebases, or
  non-PostgreSQL databases (MySQL, SQLite, etc.).

SQLAlchemy 2.0+ Patterns (Async · PostgreSQL)

The definitive reference for writing SQLAlchemy 2.0+ code. PostgreSQL is the target database. **Async is the default execution model.** Every pattern here is 2.0-style — zero legacy `Column()` / `session.query()` / `declarative_base()` patterns. When you write any model, query, or migration, follow these rules exactly.

> This is a **global, project-agnostic** skill. Examples use generic names (`User`, `Order`). Adapt names to the project, never copy project-specific session names or pool numbers from examples as if they were rules.

---

Quick Reference Cheatsheet (scan in 30 seconds)

| Rule | Do this | |---|---| | Base class | `class Base(DeclarativeBase)` — never `declarative_base()` | | Columns | `mapped_column()` + `Mapped[T]` — never bare `Column()` | | Nullable | `Mapped[str]` = NOT NULL · `Mapped[str \| None]` = NULL | | PK (uuid) | `Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid4)` | | Created/updated | `server_default=func.now()`; updated adds `onupdate=...` | | Relationship | `Mapped[list["X"]]` (collection) / `Mapped["X"]` (scalar) + `back_populates` — never `backref` | | FK | always on the **many** side; `ForeignKey("t.id", ondelete="CASCADE")` | | Async sessionmaker | `async_sessionmaker(engine, expire_on_commit=False)` — `expire_on_commit=False` is required | | Query | `select(Model)` + `await session.execute(...)` — never `session.query()` in async | | Get list of entities | `(await session.scalars(stmt)).all()` | | Get one or none | `(await session.execute(stmt)).scalar_one_or_none()` | | Relationship access | declare `selectinload` (collections) / `joinedload` (scalars) — lazy load = `MissingGreenlet` in async | | `joinedload` collection | add `.unique()` to the result — mandatory | | Eager load + filter | `selectinload(User.roles.and_(Role.active))` — **NOT** `.where()` | | List endpoint | always add `load_only(...)` with only the columns the response needs | | N+1 | never query inside a loop — `WHERE id IN (...)` once | | Bulk insert | `await session.execute(insert(Model), [{...}, ...])` — never `add()` in a loop | | Count | `await session.scalar(select(func.count()).select_from(Model))` — never `len(.all())` | | NULL test | `col.is_(None)` / `col.is_not(None)` — never `== None` | | Upsert | `pg_insert(Model)...on_conflict_do_update(index_elements=[...], set_={...})` | | JSON column | `Mapped[dict] = mapped_column(JSONB, default=dict)` — JSONB, never JSON | | Commit lives in | the service/unit-of-work layer — never in a repository helper |

---

Pre-Query Checklist (run before writing ANY DB function)

1. **Read-only?** Use the read/replica session if the project exposes one; otherwise the standard session. 2. **Returns a list?** Add `load_only(...)` selecting only the columns the response schema needs. 3. **Touches a relationship?** Add an explicit loader on the outer query — `selectinload` for collections, `joinedload` for scalars. Never lazy-load in async. 4. **Any query inside a loop?** Replace with one `WHERE col.in_([...])`. Never call `session.get()` / `session.execute()` per iteration. 5. **Counting?** Use `select(func.count())`. Never `len((await session.scalars(...)).all())`. 6. **Async?** `select()` + `await session.execute/scalars`. Never `session.query()`. Confirm `expire_on_commit=False`. 7. **Filtering on NULL?** `is_()` / `is_not()`. Never `== None`. 8. **Bulk write?** `add_all()` or Core `insert/update/delete`. Never `add()` in a loop. 9. **Insert that may collide?** PostgreSQL `insert` + `on_conflict_do_update`. 10. **Case-insensitive match?** `ilike()` or `func.lower(col) == value.lower()`. 11. **Where does `commit()` live?** Service layer. Repository functions read/stage only.

---

Section 1 — Declarative Models (2.0 style)

**Rule: subclass `DeclarativeBase`.** `declarative_base()` is the legacy 1.x factory; the class form gives full PEP 484 typing with no plugins.

# WRONG — legacy
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class User(Base):
    id = Column(Integer, primary_key=True)   # untyped, no Mapped[]

# CORRECT — 2.0
import uuid
from uuid import uuid4
from datetime import datetime
from sqlalchemy import MetaData, Uuid, func, text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

# Constraint naming convention — so Alembic generates stable, predictable names.
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",
}

class Base(DeclarativeBase):
    metadata = MetaData(naming_convention=NAMING_CONVENTION)

*Why the naming convention: unnamed constraints get random DB-assigned names; autogenerated migrations then can't reliably drop/alter them.*

**Rule: every column is `mapped_column()` + `Mapped[T]`.** `mapped_column()` reads the annotation for type and nullability. Bare `Column()` carries no ORM typing.

**Nullability is inferred from the annotation — do not also pass `nullable=`:**

class User(Base):
    _