SQLAlchemy 2.0 skill for Codex

Done well means: models use typed Mapped[...] annotations, every query is a select() executed through a session whose lifetime is one unit of work, relationships load only when the code asks (lazy loads raise instead of firing silently), and every schema change is an Alembic revision someone has read.

Inspect first

1. Read AGENTS.md, then pyproject.toml or requirements: sqlalchemy version (2.0.x or 2.1), driver (psycopg, asyncpg, pymysql, aiosqlite), alembic.
2. Find where the engine and session factory live, and how a request or job gets a session (FastAPI dependency, Flask extension, context manager).
3. Grep for legacy patterns to scope a migration: rg -n "session\.query\(|declarative_base\(|\.get\(|autocommit=|engine\.execute" --type py.
4. Read alembic.ini, migrations/env.py (or alembic/env.py) and the latest revision.

Models

from datetime import datetime
from decimal import Decimal
from sqlalchemy import ForeignKey, MetaData, String, Numeric, func, Index
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

NAMING = {
    "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)

class Customer(Base):
    __tablename__ = "customers"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(320), unique=True)
    orders: Mapped[list["Order"]] = relationship(back_populates="customer", lazy="raise")

class Order(Base):
    __tablename__ = "orders"
    __table_args__ = (Index("ix_orders_customer_created", "customer_id", "created_at"),)
    id: Mapped[int] = mapped_column(primary_key=True)
    customer_id: Mapped[int] = mapped_column(ForeignKey("customers.id", ondelete="RESTRICT"))
    total: Mapped[Decimal] = mapped_column(Numeric(12, 2))
    note: Mapped[str | None]                       # Optional -> nullable
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    customer: Mapped[Customer] = relationship(back_populates="orders", lazy="raise")

• Mapped[str] is NOT NULL; Mapped[str | None] is nullable. Let annotations drive nullability.
• Set a naming convention before the first migration; unnamed constraints make Alembic downgrades and renames painful.
• Prefer server_default for values the database should own; default= only applies to ORM inserts.
• lazy="raise" (or raiseload("*") per query) turns accidental N+1 into an error.

Engine and session scope

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine(DATABASE_URL, pool_size=5, max_overflow=5, pool_pre_ping=True)
SessionLocal = sessionmaker(engine, expire_on_commit=False)

def place_order(customer_id: int, total: Decimal) -> int:
    with SessionLocal() as session, session.begin():   # commits or rolls back
        order = Order(customer_id=customer_id, total=total)
        session.add(order)
        session.flush()            # assigns order.id inside the transaction
        return order.id

• One session per request or job; never a module-level global session shared across threads.
• session.begin() as a context manager replaces manual commit/rollback.
• expire_on_commit=False keeps loaded attributes readable after commit (required for async).

Querying the 2.0 way

Legacy (1.x)	2.0
session.query(User).filter_by(id=1).first()	session.scalars(select(User).filter_by(id=1)).first()
session.query(User).get(1)	session.get(User, 1)
query.all()	session.scalars(stmt).all()
session.query(User.id, User.email)	session.execute(select(User.id, User.email)).all() (Row tuples)
query.count()	session.scalar(select(func.count()).select_from(User))
query.update({...})	session.execute(update(User).where(...).values(...))
engine.execute(sql)	with engine.connect() as conn: conn.execute(text(sql), params)

from sqlalchemy import select
from sqlalchemy.orm import selectinload, joinedload

stmt = (
    select(Order)
    .where(Order.customer_id == cid, Order.created_at >= since)
    .options(joinedload(Order.customer), selectinload(Order.items))
    .order_by(Order.created_at.desc(), Order.id.desc())
    .limit(50)
)
orders = session.scalars(stmt).unique().all()

Eager loading choice:

Relationship	Loader
many-to-one / one-to-one	joinedload
one-to-many / many-to-many collections	selectinload (one extra IN query)
Anything not needed by this code path	raiseload / lazy="raise"
Only some columns	load_only(Order.id, Order.total)

joinedload on a collection requires .unique() on the result. Do not joinedload large collections: rows multiply.

Bulk writes and upserts

from sqlalchemy import insert
from sqlalchemy.dialects.postgresql import insert as pg_insert

session.execute(insert(Order), [{"customer_id": 1, "total": 10}, ...])  # executemany fast path

stmt = pg_insert(Customer).values(rows)
stmt = stmt.on_conflict_do_update(index_elements=[Customer.email], set_={"name": stmt.excluded.name})
session.execute(stmt)

Chunk large loads (for example 1 000-5 000 rows per execute). For very large imports use COPY through the driver.

Async

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

engine = create_async_engine("postgresql+asyncpg://...", pool_pre_ping=True)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_session():                       # FastAPI dependency
    async with SessionLocal() as session:
        yield session

async def list_orders(session: AsyncSession, cid: int) -> list[Order]:
    result = await session.scalars(
        select(Order).where(Order.customer_id == cid).options(selectinload(Order.items)))
    return list(result)

• MissingGreenlet means a lazy load or expired attribute was touched outside await. Fix by eager loading in the query or await obj.awaitable_attrs.items (mix AsyncAttrs into Base), not by wrapping in run_sync.
• One AsyncSession per task; never share it across concurrent asyncio.gather calls.
• Behind PgBouncer in transaction mode with asyncpg, disable both statement caches: ?prepared_statement_cache_size=0 on the URL (SQLAlchemy’s cache) and connect_args={"statement_cache_size": 0} (asyncpg’s). The asyncpg dialect docs cover unique statement names if errors persist. On serverless, consider poolclass=NullPool.

Alembic

alembic revision --autogenerate -m "add orders customer index"
alembic upgrade head
alembic check            # fails if models and database differ (no new revision needed)
alembic upgrade head --sql > plan.sql   # offline SQL for review

• In env.py, set target_metadata = Base.metadata and import every model module so autogenerate sees them. For async projects start from alembic init -t async.
• Autogenerate does not detect column or table renames (it emits drop + add), anonymous constraint changes, or some server default changes. Read and fix every revision.
• SQLite needs render_as_batch=True in context.configure for ALTER support.
• Large Postgres tables: create indexes with op.create_index(..., postgresql_concurrently=True) inside with op.get_context().autocommit_block():.
• Data migrations go in their own revision with batched updates; do not import ORM models into revisions (they change later); use sa.table() stubs or SQL.
• alembic downgrade on shared environments and any revision that drops columns or tables need explicit user confirmation.
• Two heads after merging branches: alembic merge heads -m "merge".

Gotchas

• session.execute(select(User)) returns Rows of one element; use scalars() for ORM objects.
• Passing an f-string to text() is SQL injection; bind with text("... :id") and params.
• DetachedInstanceError after the session closes: load what you need inside the scope or keep expire_on_commit=False.
• pool_size times worker processes must stay under the database’s connection limit.
• Decimal columns return Decimal; JSON serializers need a handler.

Verification

• Run with echo=True (or the sqlalchemy.engine logger at INFO) on the changed path and count statements; they must not grow with result size.
• alembic upgrade head on an empty database, then alembic check reports no drift.
• Test alembic downgrade -1 && alembic upgrade head locally for the new revision.
• Type-check with mypy or pyright; Mapped annotations make query results typed.

Output

**Models changed:** <classes/columns/relationships>
**Revision:** migrations/versions/<rev>_<slug>.py (autogenerate edited: <what and why>)
**Loading strategy:** <loaders added, raiseload coverage>
**Verified:** upgrade/downgrade on fresh DB, alembic check, query count, tests