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.

# 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):
    __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.

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

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:

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

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

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:

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

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:

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.

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

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

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.

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

# 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

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

# Whole entities → list[Model]
stmt = select(User).where(User.is_active.is_(True)).order

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [OmarSaleh506](https://github.com/OmarSaleh506)
- **Source:** [OmarSaleh506/skills](https://github.com/OmarSaleh506/skills)
- **License:** MIT
- **Homepage:** https://skills.sh/OmarSaleh506/skills

Install and usage instructions live in the source repository linked above.