# Sqlalchemy Patterns

> >-

- **Type:** Skill
- **Install:** `agentstack add skill-omarsaleh506-skills-sqlalchemy-patterns`
- **Verified:** Yes — security-reviewed for prompt injection and unsafe behavior
- **Seller:** [OmarSaleh506](https://agentstack.voostack.com/s/omarsaleh506)
- **Installs:** 0
- **Category:** [Cloud & Infrastructure](https://agentstack.voostack.com/c/cloud-infrastructure)
- **Latest version:** 0.1.0
- **License:** MIT
- **Upstream author:** [OmarSaleh506](https://github.com/OmarSaleh506)
- **Source:** https://github.com/OmarSaleh506/skills/tree/main/skills/sqlalchemy-patterns
- **Website:** https://skills.sh/OmarSaleh506/skills

## Install

```sh
agentstack add skill-omarsaleh506-skills-sqlalchemy-patterns
```

Requires the [AgentStack CLI](https://agentstack.voostack.com/docs/cli). Works with Claude Code, Cursor, and any MCP-compatible agent.

## About

# 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

…

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

## Pricing

- **Free** — Free

## Security capabilities

Automated source analysis of v0.1.0 — what this tool can access:

- **Network access:** no
- **Filesystem access:** no
- **Shell / process execution:** no
- **Environment & secrets:** no
- **Dynamic code execution:** no

*"Yes" means the capability is present in the source — more access means more to trust, not that it is unsafe.*


## Versions

- **0.1.0** — security scan: passed — Imported from the upstream source.

## Links

- Listing page: https://agentstack.voostack.com/l/skill-omarsaleh506-skills-sqlalchemy-patterns
- Seller: https://agentstack.voostack.com/s/omarsaleh506
- Browse the marketplace: https://agentstack.voostack.com/browse

---
Listed on AgentStack — the marketplace for AI agent skills and MCP servers. Every listing is security-reviewed. Creators keep 70%.