# Sqlalchemy Patterns

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

- Skill: `omarsaleh506/sqlalchemy-patterns` (Agent Skill)
- Install (CLI): `npx skillmds@latest add omarsaleh506/sqlalchemy-patterns`
- Raw SKILL.md: https://api.skillmd.com/api/skills/omarsaleh506/sqlalchemy-patterns/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: AI & ML
- Author: OmarSaleh506 (https://skillmd.com/u/omarsaleh506)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/omarsaleh506/sqlalchemy-patterns

---


# 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.

```python
# 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=`:**

```python
class User(Base):
    __tablename__ = "user"
    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid4)
    email: Mapped[str]              # NOT NULL
    full_name: Mapped[str | None]   # NULL
    is_active: Mapped[bool] = mapped_column(server_default=text("true"))
```
*Why: `Mapped[str]` → NOT NULL, `Mapped[str | None]` → NULL. The annotation is the single source of truth.*

**Server defaults vs Python defaults — `default=` runs in Python at INSERT; `server_default=` is emitted as DDL and runs in the database.**

```python
# WRONG — created_at set by the app clock, drifts between app servers, naive datetime
created: Mapped[datetime] = mapped_column(default=datetime.utcnow)

# CORRECT — database clock, single source of truth
created: Mapped[datetime] = mapped_column(server_default=func.now())
updated: Mapped[datetime] = mapped_column(
    server_default=func.now(),
    onupdate=func.now(),   # recomputed on every UPDATE flush
)
```
*Why: `func.now()` uses the DB clock — consistent across processes and timezone-correct (`TIMESTAMP`). `onupdate=` fires on UPDATE only.*

**Integer PK:** `id: Mapped[int] = mapped_column(primary_key=True)` (autoincrement is implicit). **UUID PK:** use the core `Uuid` type (portable; emits native `UUID` on PostgreSQL) with `default=uuid4`.

**Abstract base for shared columns** — use a mixin / `__abstract__` base, not copy-paste:

```python
class TimestampMixin:
    created: Mapped[datetime] = mapped_column(server_default=func.now())
    updated: Mapped[datetime] = mapped_column(server_default=func.now(), onupdate=func.now())

class User(TimestampMixin, Base):
    __tablename__ = "user"
    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid4)
```

**`__table_args__`** — tuple for constraints/indexes; dict for options; tuple-with-trailing-dict for both:
```python
__table_args__ = (
    UniqueConstraint("email", name="uq_user_email"),
    Index("ix_user_active", "is_active"),
    {"comment": "Application users"},   # dict MUST be last
)
```

---

## Section 2 — Relationships

**Rule: type the relationship with `Mapped[...]` and pair both sides with `back_populates`. Never use `backref`** (it hides the reverse side from type checkers and configures it implicitly).

```python
# WRONG — backref (untyped reverse side) and FK on the one side
class Parent(Base):
    __tablename__ = "parent"
    id: Mapped[int] = mapped_column(primary_key=True)
    child_id: Mapped[int] = mapped_column(ForeignKey("child.id"))   # FK on the wrong side
    children = relationship("Child", backref="parent")              # backref, no Mapped[]

# CORRECT — one-to-many. FK lives on the MANY (child) side.
class Parent(Base):
    __tablename__ = "parent"
    id: Mapped[int] = mapped_column(primary_key=True)
    children: Mapped[list["Child"]] = relationship(back_populates="parent")  # collection

class Child(Base):
    __tablename__ = "child"
    id: Mapped[int] = mapped_column(primary_key=True)
    parent_id: Mapped[int] = mapped_column(ForeignKey("parent.id", ondelete="CASCADE"))
    parent: Mapped["Parent"] = relationship(back_populates="children")       # scalar
```
*Why FK on the many side: a child has one parent; the parent's "many children" is derived from the children's FK.*

**`cascade="all, delete-orphan"`** — deletes children when the parent is deleted AND when a child is removed from the collection. Pair it with a DB-side `ondelete="CASCADE"` + `passive_deletes=True` so the database does the bulk delete:
```python
children: Mapped[list["Child"]] = relationship(
    back_populates="parent", cascade="all, delete-orphan", passive_deletes=True
)
```
*Why `passive_deletes=True`: without it, the ORM SELECTs every child then DELETEs them one by one. With it, the DB `ON DELETE CASCADE` handles them in one statement.*

**Many-to-many (no extra columns) — `secondary=` with a `Table`:**
```python
user_role = Table(
    "user_role", Base.metadata,
    Column("user_id", ForeignKey("user.id", ondelete="CASCADE"), primary_key=True),
    Column("role_id", ForeignKey("role.id", ondelete="CASCADE"), primary_key=True),
)
class User(Base):
    roles: Mapped[list["Role"]] = relationship(secondary=user_role, back_populates="users")
```

**Many-to-many WITH extra columns — association object** (map the join table as a class):
```python
class UserRole(Base):                              # the association row
    __tablename__ = "user_role"
    user_id: Mapped[int] = mapped_column(ForeignKey("user.id"), primary_key=True)
    role_id: Mapped[int] = mapped_column(ForeignKey("role.id"), primary_key=True)
    granted_at: Mapped[datetime] = mapped_column(server_default=func.now())
    user: Mapped["User"] = relationship(back_populates="role_links")
    role: Mapped["Role"] = relationship(back_populates="user_links")
```
*Why: `secondary=` can't store columns on the join. The moment the link needs its own data, use an association object.*

**`viewonly=True`** — read-only relationship (computed join, never flushed). Use for derived collections; mutations to it are silently ignored, so never write through a `viewonly` relationship.

**Self-referential (trees)** — set `remote_side` to the PK:
```python
class Node(Base):
    __tablename__ = "node"
    id: Mapped[int] = mapped_column(primary_key=True)
    parent_id: Mapped[int | None] = mapped_column(ForeignKey("node.id"))
    children: Mapped[list["Node"]] = relationship(back_populates="parent")
    parent: Mapped["Node | None"] = relationship(back_populates="children", remote_side=[id])
```

**Non-FK join — `primaryjoin`** with `foreign()`/`remote()` markers when there is no real ForeignKey (e.g. IP-range containment). Always `viewonly=True` for these.

**Catch accidental lazy loads — set `lazy="raise_on_sql"` on relationships you always eager-load.** It raises only when a load would emit SQL, surfacing missing loader options in tests before they hit production as `MissingGreenlet`.

---

## Section 3 — Type Safety with `Mapped[]`

**Rule: every column is typed; mypy/pyright understand `Mapped[T]` natively in 2.0 — no stubs.**

```python
# WRONG — untyped column (no Mapped[T]) and a shared mutable default
class Account(Base):
    id = mapped_column(Integer, primary_key=True)   # no Mapped[] → no type checking
    meta: Mapped[dict] = mapped_column(JSONB, default={})   # {} shared across ALL rows

# CORRECT
import enum
import uuid
from sqlalchemy import Enum, String, Uuid
from sqlalchemy.dialects.postgresql import ARRAY, JSONB

class Status(enum.Enum):
    ACTIVE = "active"
    BANNED = "banned"

class Account(Base):
    __tablename__ = "account"
    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
    status: Mapped[Status] = mapped_column(Enum(Status, name="account_status"))
    tags: Mapped[list[str]] = mapped_column(ARRAY(String))
    meta: Mapped[dict] = mapped_column(JSONB, default=dict)   # default=dict → fresh {} per row
```
*Why `default=dict` (not `default={}`): a literal `{}` is shared across all instances — a classic mutable-default bug.*

**`type_annotation_map` — map a Python type to a SQL type once, project-wide:**
```python
from typing import Annotated
from sqlalchemy.orm import registry

str_255 = Annotated[str, 255]

class Base(DeclarativeBase):
    registry = registry(type_annotation_map={str_255: String(255), dict: JSONB})

class User(Base):
    name: Mapped[str_255]   # → String(255)
    meta: Mapped[dict]      # → JSONB
```

**Custom domain type — `TypeDecorator` (always set `cache_ok`):**
```python
from sqlalchemy import types

class LowerString(types.TypeDecorator):
    impl = types.String
    cache_ok = True   # REQUIRED — unset disables statement caching and warns
    def process_bind_param(self, value, dialect):
        return value.lower() if value is not None else value
```
*Why `cache_ok = True`: SQLAlchemy 2.0 caches compiled statements keyed by type. An unset `cache_ok` emits a warning and silently disables caching for every statement using the type — a real perf regression.*

**`column_property()` for a read-only computed column** — see Section 14.

---

## Section 4 — Async Session Lifecycle

**Rule: `async_sessionmaker(..., expire_on_commit=False)`.** This is non-negotiable in async.

```python
# WRONG — default expire_on_commit=True
async_session = async_sessionmaker(engine)
async with async_session() as s:
    user = await s.get(User, uid)
    await s.commit()
    return user.email   # 💥 MissingGreenlet — attribute expired, reload needs IO

# CORRECT
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

engine = create_async_engine(
    "postgresql+asyncpg://user:pw@host/db",
    pool_pre_ping=True,     # detect dead connections before use
    echo=False,             # never True in production — logs every statement
)
async_session = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
```
*Why `expire_on_commit=False`: after commit, the default expires all attributes; the next attribute access triggers a synchronous reload, which has no greenlet in async → `MissingGreenlet`. Disabling it keeps loaded data usable after commit.*

**Session scope — always a context manager. Two transaction styles:**
```python
# Auto-commit on success, auto-rollback on exception:
async with async_session() as session, session.begin():
    session.add(obj)
# (no explicit commit — begin() commits at block exit)

# Manual control:
async with async_session() as session:
    session.add(obj)
    await session.commit()
```
*Why `async with`: guarantees the connection returns to the pool even on exception. Manual `close()` leaks on error paths.*

**Read replica — two engines, two sessionmakers.** Route read-only handlers to the replica sessionmaker, writes to the primary. (The exact dependency/session names are project-specific — use whatever the project exposes.)

**`session.get(Model, pk)`** checks the identity map first and emits a SELECT only on a miss — the right tool for a single PK lookup. **Never `session.query()` in async** — it's the legacy sync API and unsupported on `AsyncSession`.

**Need one lazy attribute without an upfront loader?** Use `AsyncAttrs` (2.0.13+): `class Base(AsyncAttrs, DeclarativeBase): ...` then `await obj.awaitable_attrs.children`. Prefer eager loaders for anything in a hot path.

**`AsyncSession` is NOT concurrency-safe** — never share one session across `asyncio.gather` tasks. One session per task.

**`run_sync()`** — only for sync-only operations like DDL: `await conn.run_sync(Base.metadata.create_all)`.

---

## Section 5 — Querying: Complete SELECT Guide

```python
from sqlalchemy import select, and_, or_, not_, nulls_last

# Whole entities → list[Model]
stmt = select(User).where(User.is_active.is_(True)).order_by(User.created.desc())
users = (await session.scalars(stmt)).all()

# Specific columns → list[Row]
stmt = select(User.id, User.email)
rows = (await session.execute(stmt)).all()   # each row: row.id, row.email
```

**WHERE composition:** multiple `.where(a, b)` args or `and_(a, b)` = AND; `or_(...)`; `not_(...)` / `~`.

**Ordering:** `User.col.desc()`, `User.col.asc()`, `nulls_last(User.col)`. **Pagination:** `.limit(n).offset(m)`. **Distinct:** `.distinct()`, or PostgreSQL `DISTINCT ON`: `select(User).distinct(User.email)`.

**Terminal methods — picking wrong returns the wrong shape:**

| Call | Returns |
|---|---|
| `(await session.scalars(stmt)).all()` | `list[Model]` (ORM objects) |
| `await session.scalar(stmt)` | first column of first row, or `None` |
| `(await session.execute(stmt)).scalar_one()` | exactly one value; raises if 0 or >1 |
| `(await session.execute(stmt)).scalar_one_or_none()` | one or `None`; raises if >1 |
| `(await session.execute(stmt)).first()` | first `Row` or `None` |
| `(await session.execute(stmt)).all()` | `list[Row]` (tuples) |

*Why `scalar_one_or_none()` for "fetch by unique key": it asserts uniqueness — a duplicate raises instead of silently returning the first.*

**Filtering operators:**
```python
# WRONG — Python truthiness / == on NULL, and filter_by for a complex predicate
select(User).where(User.deleted_at == None)        # linters flag this; fragile
select(User).filter_by(age > 18)                   # filter_by takes kwargs only → TypeError

# CORRECT
select(User).where(User.deleted_at.is_(None))      # emits IS NULL
select(User).where(User.age > 18)                  # use .where() for anything but simple equality

User.id.in_(ids)                       # IN
User.id.not_in(ids)                    # NOT IN
User.email.ilike("%@gmail.com")        # case-insensitive LIKE
User.age.between(18, 65)
User.deleted_at.is_(None)              # IS NULL  — NEVER == None
User.deleted_at.is_not(None)           # IS NOT NULL
func.lower(User.email) == email.lower()  # case-insensitive equality

from sqlalchemy import exists
stmt = select(exists().where(User.email == email))   # → await session.scalar(stmt) → bool
```
*Why `is_(None)` not `== None`: `== None` works by luck via operator overloading but linters flag it and it breaks on some expressions. `is_()` always emits `IS NULL`.*

---

## Section 6 — Relationship Loading (Mandatory in Async)

**This is the most important section. In async, accessing an unloaded relationship raises `MissingGreenlet` and kills the request.** Every relationship you touch must have an explicit loader on the outer query.

```python
from sqlalchemy.orm import selectinload, joinedload, raiseload, contains_eager

# WRONG — lazy load in async
users = (await session.scalars(select(User))).all()
for u in users:
    print(u.roles)   # 💥 MissingGreenlet

# CORRECT — collection → selectinload (second IN query)
stmt = select(User).options(selectinload(User.roles))
users = (await session.scalars(stmt)).all()

# CORRECT — scalar (many-to-one) → joinedload (LEFT OUTER JOIN, same query)
stmt = select(Order).options(joinedload(Order.customer))
orders = (await session.scalars(stmt)).all()
```

**Strategy choice:**
- **`selectinload(Model.rel)`** — collections (one-to-many, many-to-many). Emits one extra `WHERE IN` query. *Why: avoids the row-multiplication a JOIN causes on collections.*
- **`joinedload(Model.rel)`** — scalars (many-to-one, one-to-one). Adds a `LEFT OUTER JOIN`. Use `joinedload(Order.customer, innerjoin=True)` when the FK is NOT NULL.
- **`subqueryload` / `lazyload`** — never in async.
- **`raiseload("*")`** — add in development/tests to turn any accidental lazy load into an immediate error.

**`joinedload` on a collection requires `.unique()`** on the result — otherwise the JOIN's duplicate rows produce duplicate parents (and 2.0 raises):
```python
stmt = select(User).options(joinedload(User.roles))   # collection via JOIN
users = (await session.scalars(stmt)).unique().all()  # .unique() MANDATORY
```
(Prefer `selectinload` for collections — no `.unique()` needed.)

**Nesting:**
```python
select(User).options(selectinload(User.roles).selectinload(Role.permissions))
```

**Eager load WITH a filter — use `relationship.and_()`, NOT `.where()`:**
```python
# WRONG — Load options have no .where(); this is invalid syntax
select(User).options(selectinload(User.roles).where(Role.is_active == True))

# CORRECT — put the criteria inside the relationship attribute
select(User).options(selectinload(User.roles.and_(Role.is_active == True)))
```
*Why: the criteria attaches to the relationship's join condition. `.and_()` works identically across `selectinload`/`joinedload`/`lazyload`.*

**Global per-entity criteria — `with_loader_criteria()`** (applies everywhere that entity loads, including nested loads — ideal for soft-delete):
```python
from sqlalchemy.orm import with_loader_criteria
select(User).options(
    selectinload(User.roles),
    with_loader_criteria(Role, lambda cls: cls.deleted_at.is_(None)),
)
```

**`contains_eager(Model.rel)`** — when you've already written an explicit `.join()` and want it to populate the relationship (e.g. to filter on the joined table):
```python
stmt = (select(User).join(User.roles)
        .where(Role.name == "admin")
        .options(contains_eager(User.roles)))
```

---

## Section 7 — Partial Loading: `load_only` and `defer`

**Rule: every list endpoint uses `load_only()` with only the columns the response schema returns.** Large tables have heavy columns (blobs, JSONB, text) you don't need in a list view.

```python
from sqlalchemy.orm import load_only, defer, undefer

# WRONG — list endpoint pulls every column, including heavy text/JSONB/blob columns
stmt = select(User)

# CORRECT — list view fetches only what the response needs
stmt = select(User).options(load_only(User.id, User.email, User.is_active, User.created))
```
*Why: `SELECT *` on a wide table moves megabytes per page you immediately discard.*

**`defer(col)`** — exclude one heavy column, load it lazily only if accessed. **`undefer(col)`** — force-load a column deferred at the mapping level.

**Combine column + relationship loaders in one `.options()`:**
```python
stmt = select(User).options(
    load_only(User.id, User.email),
    selectinload(User.roles).load_only(Role.id, Role.name),
)
```

**Raise instead of lazy-loading a column** — `load_only(..., raiseload=True)` (2.0) whitelists columns and raises on access to any other; `defer(col, raiseload=True)` (1.4) raises on access to that one column. Use to catch unintended column loads in tests.

---

## Section 8 — N+1 Elimination & Batch Patterns

**N+1 = one query to get N rows, then one query per row inside a loop. Never do this.**

```python
# WRONG — N+1 (one query per id)
for uid in user_ids:
    user = await session.get(User, uid)

# CORRECT — one query for all
users = (await session.scalars(
    select(User).where(User.id.in_(user_ids))
)).all()
```
*Why: N round-trips to the DB dominate latency; one `IN` query is a single round-trip.*

**For relationship N+1 → `selectinload`** (Section 6). **For repeated single-PK access of already-loaded objects → `session.get()`** uses the identity map (no SQL).

**Bulk writes — never `add()` in a loop:**
```python
# WRONG
for row in rows:
    session.add(User(**row))   # unit-of-work overhead per object

# CORRECT — ORM, one flush
session.add_all([User(**row) for row in rows])
await session.commit()

# FASTEST — Core bulk insert (list of dicts as 2nd arg, not .values())
from sqlalchemy import insert
await session.execute(insert(User), [{"email": e} for e in emails])
await session.commit()
```

**Bulk UPDATE / DELETE — one statement, not a loop:**
```python
from sqlalchemy import update, delete
await session.execute(update(User).where(User.id.in_(ids)).values(is_active=False))
await session.execute(delete(User).where(User.id.in_(ids)))
```
*Note: ORM-enabled bulk UPDATE/DELETE use `synchronize_session` (default `"auto"`) to keep in-session objects consistent — leave it at `"auto"` unless profiling says otherwise.*

---

## Section 9 — Counting & Aggregation

**Rule: count in the database. Never fetch rows to count them.**

```python
from sqlalchemy import func

# WRONG — loads every row into memory just to count
n = len((await session.scalars(select(User).where(User.is_active.is_(True)))).all())

# CORRECT
n = await session.scalar(
    select(func.count()).select_from(User).where(User.is_active.is_(True))
)
```
*Why: `len(.all())` transfers the whole result set; `COUNT(*)` returns a single integer.*

**Other aggregates — same pattern:** `func.sum(col)`, `func.avg(col)`, `func.max(col)`, `func.min(col)`.

**Group / having / conditional count (PostgreSQL `FILTER`):**
```python
stmt = (
    select(User.team_id,
           func.count().label("total"),
           func.count().filter(User.is_active.is_(True)).label("active"))
    .group_by(User.team_id)
    .having(func.count() > 5)
)
rows = (await session.execute(stmt)).all()
```

---

## Section 10 — Subqueries, EXISTS, CTEs

```python
from sqlalchemy import exists, select, func, union_all

# WRONG — IN (subquery) materializes the entire id set just to test membership
select(User).where(User.id.in_(select(Order.user_id).where(Order.total > 100)))  # for "has any", prefer EXISTS

# CORRECT — EXISTS short-circuits on the first matching row
select(User).where(select(1).where(Order.user_id == User.id, Order.total > 100).exists())

# Scalar subquery as a value
order_count = (
    select(func.count(Order.id)).where(Order.user_id == User.id).scalar_subquery()
)

# EXISTS — prefer over IN for "has any" checks
has_admin = await session.scalar(select(exists().where(Role.name == "admin")))

# Relationship EXISTS helpers
select(User).where(User.roles.any(Role.name == "admin"))   # collection: any()
select(Order).where(Order.customer.has(User.is_active.is_(True)))  # scalar: has()
```
*Why `exists()` over `IN (subquery)`: the planner short-circuits on the first match instead of materializing the full set.*

**Recursive CTE (tree expansion) — always include a depth guard / `UNION ALL`:**
```python
# Base term: roots
base = select(Node.id, Node.parent_id, literal(0).label("depth")).where(Node.parent_id.is_(None))
cte = base.cte("tree", recursive=True)
parent = cte.alias()
# Recursive term: children of already-collected nodes, with a depth cap
recursive = (
    select(Node.id, Node.parent_id, (parent.c.depth + 1).label("depth"))
    .join(parent, Node.parent_id == parent.c.id)
    .where(parent.c.depth < 10)   # guard against cycles / runaway recursion
)
cte = cte.union_all(recursive)
rows = (await session.execute(select(cte.c.id, cte.c.depth))).all()
```
*Why the depth guard: a cycle in self-referential data makes a recursive CTE loop forever.*

---

## Section 11 — Transactions

**Rule: the unit of work owns the transaction. Use `session.begin()` in the service layer; repository helpers stage work and never commit.**

```python
# WRONG — repository helper commits; caller can no longer roll back the whole unit of work
async def create_user(session, data):
    user = User(**data)
    session.add(user)
    await session.commit()              # 💥 partial commit — if assign_role later fails, user persists

# CORRECT — service layer owns the transaction boundary; repos only stage work
async with async_session() as session:
    async with session.begin():         # commit on success, rollback on exception
        await create_user(session, data)
        await assign_role(session, user_id, role_id)
# both succeed or both roll back
```
*Why: if a repository commits, a later step's failure can't roll back the earlier write — you get partial, inconsistent state.*

- **`await session.flush()`** — push pending changes to the DB (assigns PKs, runs constraints) without committing. Use when you need a generated PK mid-transaction.
- **`await session.commit()`** — flush + COMMIT. **`await session.rollback()`** — discard everything since the last commit.
- **`await session.refresh(obj)`** — reload from DB (needed only when `expire_on_commit=True`).
- **Savepoints — `async with session.begin_nested():`** — partial rollback inside a larger transaction (e.g. retrying one row in a batch without aborting the rest).

---

## Section 12 — PostgreSQL-Specific Patterns

**Types** (import from `sqlalchemy.dialects.postgresql as pg`):
```python
from sqlalchemy import JSON, Uuid
from sqlalchemy.dialects import postgresql as pg

# WRONG — plain JSON (text storage, not indexable, no containment operators)
data: Mapped[dict] = mapped_column(JSON, default=dict)

# CORRECT
id:    Mapped[uuid.UUID] = mapped_column(Uuid, default=uuid4)        # portable, native UUID on PG
data:  Mapped[dict]      = mapped_column(pg.JSONB, default=dict)     # ALWAYS JSONB, never JSON
tags:  Mapped[list[str]] = mapped_column(pg.ARRAY(String))
ip:    Mapped[str]       = mapped_column(pg.INET)
status: Mapped[str]      = mapped_column(pg.ENUM("a", "b", name="my_enum", create_type=False))
```
*Why JSONB over JSON: JSONB is binary, indexable (GIN), and supports containment operators. Plain JSON is text — no indexing. Why `create_type=False`: lets Alembic manage the `CREATE TYPE` so the migration doesn't try to recreate it.*
*`pg.UUID(as_uuid=True)` is the PG-specific equivalent of `Uuid` — use core `Uuid` unless you need PG-only behavior.*

**JSONB operations** (operators verified against the PG dialect):
```python
Model.data.op("->>")("key")        # -> 'key'::text   (text value)
Model.data.op("->")("key")         # -> 'key'         (json value)
Model.data.contains({"k": "v"})    # @>  containment
Model.data.has_key("k")            # ?   key exists
Model.data.has_any(["a", "b"])     # ?|  any key exists
func.jsonb_set(Model.data, "{k}", '"v"')   # update nested key
```

**Upsert — `INSERT ... ON CONFLICT`** (must import `insert` from the PG dialect):
```python
from sqlalchemy.dialects.postgresql import insert as pg_insert

stmt = pg_insert(User).values([{"email": e, "name": n} for e, n in rows])
stmt = stmt.on_conflict_do_update(
    index_elements=[User.email],                       # the unique key
    set_={"name": stmt.excluded.name, "updated": func.now()},  # excluded = the row that failed to insert
)
await session.execute(stmt)
# Ignore-on-conflict variant:
await session.execute(pg_insert(User).values(...).on_conflict_do_nothing(index_elements=[User.email]))
```

**RETURNING:**
```python
stmt = pg_insert(User).values(email="a@b.c").returning(User.id, User.created)
row = (await session.execute(stmt)).one()
```

**Window functions:**
```python
func.row_number().over(partition_by=Order.user_id, order_by=Order.created.desc())
func.rank().over(order_by=Order.total.desc())
func.lag(Order.total).over(order_by=Order.created)
```

**Full-text search:**
```python
# Query
stmt = select(Doc).where(
    func.to_tsvector("english", Doc.body).op("@@")(func.plainto_tsquery("english", q))
)
# Index (GIN on the tsvector expression)
Index("ix_doc_fts", func.to_tsvector("english", Doc.body), postgresql_using="gin")
```

**Common `func.*`:** `func.now()`, `func.gen_random_uuid()`, `func.coalesce(col, default)`, `func.nullif(col, "")`, `func.array_agg(col)`, `func.string_agg(col, ", ")`.

---

## Section 13 — Indexes & Constraints

All in `__table_args__`:
```python
from sqlalchemy import Index, UniqueConstraint, CheckConstraint, ForeignKeyConstraint, func

# WRONG — plain index on email, but queries filter func.lower(email): the index is never used
#   Index("ix_user_email", "email")  + WHERE func.lower(email) == x   → sequential scan
#   Index("ix_user_meta_btree", "meta")  → B-tree can't serve JSONB containment (@>)

# CORRECT
__table_args__ = (
    Index("ix_user_email_lower", func.lower(User.email)),                  # functional
    Index("ix_user_meta", "meta", postgresql_using="gin"),                 # GIN for JSONB/ARRAY/FTS
    Index("ix_user_active", "team_id",
          postgresql_where=text("is_active")),                            # partial index
    Index("ix_order_cover", "user_id", postgresql_include=["status"]),     # covering index
    UniqueConstraint("email", name="uq_user_email"),
    CheckConstraint("age >= 0", name="ck_user_age_nonneg"),
)
```
*Why functional index `lower(email)`: a query filtering `func.lower(email) == x` can only use an index built on `lower(email)`. Why GIN: B-tree can't index JSONB/array containment. Why partial: a `WHERE is_active` index is smaller and faster when you only ever query active rows.*

For multi-column FKs use `ForeignKeyConstraint([...], [...], ondelete="CASCADE", name="fk_...")`.

---

## Section 14 — Hybrid & Column Properties

**`@hybrid_property`** — one definition that works as a Python attribute AND in SQL `WHERE`. **Use the `.inplace.expression` form (2.0.4+)** so type checkers stay happy:
```python
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy import ColumnElement, func, type_coerce, Float

class Interval(Base):
    start: Mapped[int]
    end: Mapped[int]

    @hybrid_property
    def length(self) -> int:               # Python side
        return self.end - self.start

    # WRONG — redefining `length` as the expression trips PEP 484 type checkers
    # @length.expression
    # def length(cls):
    #     return cls.end - cls.start

    # CORRECT — .inplace lets the SQL method keep a private name (2.0.4+)
    @length.inplace.expression
    @classmethod
    def _length_expr(cls) -> ColumnElement[int]:   # SQL side
        return cls.end - cls.start
```
*Why `.inplace`: the old `@length.expression` redefines the same name and trips PEP 484 checkers. `.inplace` mutates the hybrid in place so the SQL method can have a private name.*

**`column_property()`** — a read-only column computed from a subquery, loaded with the entity:
```python
from sqlalchemy.orm import column_property
order_count: Mapped[int] = column_property(
    select(func.count(Order.id)).where(Order.user_id == id).correlate_except(Order).scalar_subquery()
)
```

**`association_proxy`** — expose a nested attribute directly (e.g. `user.role_names` from `user.roles[].name`).

---

## Section 15 — Events

**Events are synchronous.** Inside a handler: no `await`, no lazy loads, no ORM queries needing the greenlet. Emit SQL only through the passed `connection`.

```python
from sqlalchemy import event
from sqlalchemy.orm import Session

# WRONG — events are synchronous; awaiting or lazy-loading inside one fails
@event.listens_for(Model, "before_insert")
async def bad(mapper, connection, target):     # handler can't be a coroutine
    target.owner = await fetch_owner(target.owner_id)   # 💥 no greenlet here

# CORRECT — set local attributes only; emit SQL via the passed connection
@event.listens_for(Model, "before_insert")
def set_defaults(mapper, connection, target):
    target.slug = slugify(target.name)

@event.listens_for(Session, "before_flush")
def audit(session, flush_context, instances):
    for obj in session.new:
        ...   # inspect pending objects
```

**In async, attach session events to the sync session class:**
```python
@event.listens_for(AsyncSession.sync_session_class, "before_flush")
def _before_flush(session, ctx, instances):
    ...
```

**`@event.listens_for(Engine, "connect")`** — set per-connection options (e.g. statement timeout, timezone) at connect time.

---

## Section 16 — Testing with Async SQLAlchemy

**Rule: never mock the session. Use a real database (test PostgreSQL, or SQLite for pure-ORM unit tests).** Wrap each test in a transaction and roll back — no real commits, full isolation.

```python
import pytest
from sqlalchemy.ext.asyncio import AsyncSession

# WRONG — mocking the session tests your mocks, not your SQL
def test_create_user():
    session = MagicMock()
    session.scalar.return_value = User(id=1)   # never exercises real query/constraints

# CORRECT — real DB, transaction per test, rolled back for isolation
@pytest.fixture
async def session(engine):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    async with AsyncSession(engine, expire_on_commit=False) as s:
        yield s
        await s.rollback()
```
*Why real DB: mocked sessions test your mocks, not your SQL — they miss constraint violations, type coercion, and N+1s. Why rollback per test: deterministic, isolated, no cross-test state.*

**`pyproject.toml`:** `[tool.pytest.ini_options]` → `asyncio_mode = "auto"`. Build test objects with factories (e.g. `factory_boy`), not hand-written dicts repeated across tests.

---

## Section 17 — Connection Pool & Engine Configuration

```python
# WRONG — logs every statement in prod, and stale pooled connections fail requests
engine = create_async_engine(DATABASE_URL, echo=True)   # no pre_ping, no recycle

# CORRECT
engine = create_async_engine(
    DATABASE_URL,
    pool_size=10,         # persistent connections kept open
    max_overflow=20,      # extra connections allowed under burst load
    pool_timeout=30,      # seconds to wait for a free connection before erroring
    pool_recycle=1800,    # recycle connections older than 30 min (avoids server-side timeouts)
    pool_pre_ping=True,   # test a connection's liveness before handing it out
    echo=False,           # True only in local dev
)
```
*Why `pool_pre_ping`: a DB restart or idle timeout leaves dead connections in the pool; pre-ping replaces them transparently instead of failing the request. Why `pool_recycle`: prevents the DB from closing a connection mid-request after its idle timeout.*

**Serverless / Lambda / external pooler (pgbouncer):** use `poolclass=NullPool` — no persistent pool; create and drop per request. *Why: a long-lived pool is useless when the runtime freezes between invocations, and double-pooling behind pgbouncer causes connection storms.*

---

## Section 18 — Alembic Integration

**Rule: never hand-write migration bodies — always autogenerate, then review.**
```bash
# WRONG — empty hand-written revision; drifts from the models, misses constraints/types
alembic revision -m "add is_active"      # then manually typing op.add_column(...)

# CORRECT — autogenerate from model metadata, then read the diff before applying
alembic revision --autogenerate -m "add user.is_active"
```

```python
# WRONG — autogenerate emits drop+add for a rename → silently destroys the column's data
op.drop_column("user", "fullname")
op.add_column("user", sa.Column("full_name", sa.String()))

# CORRECT — rename preserves data
op.alter_column("user", "fullname", new_column_name="full_name")
```

- In `env.py`: `target_metadata = Base.metadata` so autogenerate sees your models.
- **Always read the generated migration** — autogenerate misses column renames (it drops + recreates, losing data), some constraint changes, and server-default changes. Fix renames manually: `op.alter_column("user", "old", new_column_name="new")`.
- **PostgreSQL ENUMs:** declare `pg.ENUM(..., create_type=False)` on the model and let the migration own `CREATE TYPE` / `DROP TYPE` via `op.execute(...)`, so re-running doesn't error on an existing type.
- **Data migrations:** `op.execute("UPDATE ...")` in the migration body.
- `include_schemas=True` in the context config for multi-schema databases.
- Batch mode is for SQLite only — not needed for PostgreSQL.

---

## Troubleshooting

| Symptom | Cause | Fix |
|---|---|---|
| `MissingGreenlet: greenlet_spawn has not been called` | Lazy load / expired-attribute access in async | Add `selectinload`/`joinedload`; set `expire_on_commit=False` |
| `InvalidRequestError: ... can't be used with ... joined eager load` (collections) | `joinedload` on a collection without dedupe | Add `.unique()` to the result, or use `selectinload` |
| `selectinload(...).where(...)` AttributeError / no-op | Load options have no `.where()` | Use `selectinload(Model.rel.and_(...))` |
| `SAWarning: TypeDecorator ... will not produce a cache key` | `cache_ok` u

…(truncated)