/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)

Rules for writing SQLAlchemy 2.0 / 2.1 code where **PostgreSQL is the database and async is the execution model**. Everything here is 2.0-style — no legacy `Column()` class attributes, `session.query()`, or `declarative_base()`.

> Project-agnostic: examples use generic names (`User`, `Order`). Adapt names to the project; never copy example session names or pool numbers as if they were rules.

**Versions:** works on SQLAlchemy 2.0 and 2.1 (current release line). On 2.1: Python ≥ 3.11, install `sqlalchemy[asyncio]` (greenlet is no longer pulled in by default), and always name the driver in the URL (`postgresql+asyncpg://` / `postgresql+psycopg://`). Upgrade notes: `references/migration-2.1.md`.

---

Quick Reference Cheatsheet

| Rule | Do this | |---|---| | Base class | `class Base(AsyncAttrs, DeclarativeBase)` — never `declarative_base()` | | Columns | `mapped_column()` + `Mapped[T]` — never bare `Column()` on a mapped class | | Nullable | `Mapped[str]` = NOT NULL · `Mapped[str \| None]` = NULL | | UUID PK | `Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid4)` (maps to `Uuid` automatically) | | Timestamps | `type_annotation_map = {datetime.datetime: DateTime(timezone=True)}` on Base; `server_default=func.now()` | | `updated` column in async | `onupdate=func.now(), server_onupdate=FetchedValue()` + `__mapper_args__ = {"eager_defaults": True}` | | Relationship | `Mapped[list["X"]]` / `Mapped["X"]` + `back_populates` — never `backref` | | Cascade in async | `"save-update, merge, expunge, delete, delete-orphan"` — not `"all"` | | Huge collection | `WriteOnlyMapped["X"]` — never `DynamicMapped` (not async-compatible) | | FK | on the **many** side; `ForeignKey("t.id", ondelete="CASCADE")` | | Sessionmaker | `async_sessionmaker(engine, expire_on_commit=False)` | | Query | `select(Model)` + `await session.execute / scalars / scalar / get` | | List of entities | `(await session.scalars(stmt)).all()` | | One or none | `(await session.execute(stmt)).scalar_one_or_none()` | | Relationship access | `selectinload` (collections) / `joinedload` (scalars) on the query | | `joinedload` on a collection | add `.unique()` to the result | | Eager load + filter | `selectinload(User.roles.and_(Role.active))` — not `.where()` | | List endpoint | `load_only(...)` with only the columns the response needs | | N+1 | never query inside a loop — one `WHERE id IN (...)` | | Bulk insert | `await session.execute(insert(Model), [{...}, ...])`; `.returning(Model)` via `session.scalars` | | Count | `await session.scalar(select(func.count()).select_from(Model))` | | 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 a repository helper | | Concurrency | one `AsyncSession` per task — never shared across `asyncio.gather` |

---

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

1. **Read-only?** Use the read/replica session if the project exposes one. 2. **Returns a list?** Add `load_only(...)` with 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. Lazy loading in async raises `MissingGreenlet`. 4. **Any query inside a loop?** Replace with one `WHERE col.in_([...])`. 5. **Counting?** `select(func.count())`, never `len(... .all())`. 6. **Async correctness?** `select()` + `await session.*`; `expire_on_commit=False`; no attribute access that could trigger IO after commit or after an UPDATE flush. 7. **Filtering on NULL?** `is_()` / `is_not()`. 8. **Bulk write?** `add_all()` or Core `insert/update/delete` with a list of dicts — never `add()` in a loop. 9. **Insert that may collide?** PostgreSQL `insert` + `on_conflict_do_update` / `do_nothing`. 10. **Case-insensitive match?** `ilike()` or `func.lower(col) == value.lower()` (with a matching functional index). 11. **Where does `commit()` live?** Service layer. Repository functions read / stage / flush only.

---

Decision Guide

**Which loader?**

| You will access… | Use | |---|---| | a collection (one-to-many, many-to-many) | `selectinload(Parent.children)` | | a scalar (many-to-one, one-to-one) | `joinedload(Child.parent)` (`innerjoin=True` if FK NOT NULL) | | a collection, filtered on the related table in WHERE | explicit `.join()` + `contains_eager()` | | a subset of a collection | `selectinload(Parent.children.and_(...))` | | a huge collection | `WriteOnlyMapped` + `parent.children.select()` | | one attribute, once, no loader planned | `await obj.awaitable_attrs.rel` or `await session.refresh(obj, ["rel"])` | | nothing beyond what you loaded (tests) | add `raiseload("*")` to fail loudly |

**Which write path?**

| Situation | Use | |---|---| | A few objects; need events, relationships, cascades | `session.add()` / `add_all()` + flush | | Many rows from dicts | `session.execute(insert(Model), rows)` | | Many rows and you need them back