git:20260629.89b93c6 to git:20261006.98751eb

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

- | Base class | `class Base(DeclarativeBase)` — never `declarative_base()` |

- | Columns | `mapped_column()` + `Mapped[T]` — never bare `Column()` |

- | 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())` |

- | Commit lives in | the service/unit-of-work layer — never in a repository helper |

- 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

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

- class Base(DeclarativeBase):

- ```

- *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

- 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"

- 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` unset on a custom type | Set `cache_ok = True` |

- | `DetachedInstanceError` | Accessing attributes after the session closed | Load needed data inside the session, or `expire_on_commit=False` |

- | Duplicate rows from a query | Cartesian product from `joinedload` collection | `.unique()` or `selectinload` |

- | Enum `DuplicateObject` on migration | ENUM type recreated | `create_type=False` + let Alembic manage `CREATE TYPE` |

- | Connection errors after idle | Stale pooled connection | `pool_pre_ping=True` + `pool_recycle` |

- - **Discover a pattern not covered here while working?** Add it to the relevant section of this skill's `SKILL.md` — edit your local copy, and open a PR to the source repo so everyone benefits.

- - **User hits a SQLAlchemy error not listed?** Add the symptom + cause + fix to the Troubleshooting table.

- - **Upgrading SQLAlchemy?** Check https://docs.sqlalchemy.org/en/20/changelog/ — minor 2.0.x releases add patterns (e.g. `load_only(raiseload=)` in 2.0, `AsyncAttrs` in 2.0.13, hybrid `.inplace` in 2.0.4).

- - **When updating: only add; never remove** unless the changelog explicitly deprecates something.

- - **After any update**, append the date and what changed below.

- - 2026-06-17 — Initial version. Verified against SQLAlchemy 2.0 docs (release line 2.0.51). Covers Sections 1–18, cheatsheet, pre-query checklist, troubleshooting. Key verified corrections vs common mistakes: eager-load filtering uses `relationship.and_()` not `.where()`; `joinedload` collections require `.unique()`; core `Uuid` preferred over `pg.UUID`; `TypeDecorator.cache_ok=True` required; hybrid `.inplace.expression` for type-checker compliance.

- - 2026-06-28 — Packaged into the `omar-skills` marketplace: reworded the "Keeping This Skill Current" path to be install-agnostic, tightened the description triggers to stay SQLAlchemy-specific, and removed a contradictory `IN (subquery)` example in §10.