# SQLAlchemy 2.0 & Alembic Cheat Sheet Essentials for day-to-day ORM work. SQLAlchemy 2.0 style (`select()`, `Mapped[]`). --- ## Engine & session ```python from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker engine = create_engine( # one per process, holds the pool "postgresql+psycopg://user:pw@localhost/app", echo=False, # True logs every SQL statement — best debugging tool here pool_size=5, pool_pre_ping=True, # drop dead connections instead of erroring on first use ) SessionLocal = sessionmaker(engine, expire_on_commit=False) with Session(engine) as session: # closes (and rolls back) on exit session.add(obj) session.commit() with Session(engine) as session, session.begin(): # commit on success, session.add(obj) # rollback on exception ``` - One session per request / task / unit of work. A `Session` is **not** thread-safe and must never be shared or made global. - The session is a cache (identity map): the same row fetched twice in one session gives you the same Python object. - `commit()` expires every loaded object by default, so the next attribute access re-SELECTs. `expire_on_commit=False` avoids that, at the cost of possibly stale data. - Async: `create_async_engine` / `AsyncSession` mirror this API with `await` (`await session.execute(...)`, `await session.commit()`). ## Models ```python from datetime import datetime from sqlalchemy import ForeignKey, String, UniqueConstraint, func from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship class Base(DeclarativeBase): pass class Author(Base): __tablename__ = "authors" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String(100), index=True) bio: Mapped[str | None] # Optional → NULL allowed created_at: Mapped[datetime] = mapped_column(server_default=func.now()) books: Mapped[list["Book"]] = relationship( back_populates="author", cascade="all, delete-orphan" ) __table_args__ = (UniqueConstraint("name", name="uq_authors_name"),) class Book(Base): __tablename__ = "books" id: Mapped[int] = mapped_column(primary_key=True) title: Mapped[str] author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"), index=True) author: Mapped["Author"] = relationship(back_populates="books") ``` - The annotation drives the column: `Mapped[str]` is NOT NULL, `Mapped[str | None]` is nullable. `mapped_column(...)` only adds what the type can't express. - Always index your foreign keys — most databases don't do it for you. - `default=` is computed in Python, `server_default=` is DDL the database applies (and the only one that fills rows inserted outside your app). - `back_populates` on both sides keeps the two directions in sync in memory. ## Relationships ```python # one-to-many → the "many" side owns the FK class Author(Base): books: Mapped[list["Book"]] = relationship(back_populates="author") class Book(Base): author_id: Mapped[int] = mapped_column(ForeignKey("authors.id")) author: Mapped["Author"] = relationship(back_populates="books") # one-to-one → same thing with uselist=False on the collection side profile: Mapped["Profile"] = relationship(back_populates="author", uselist=False) # many-to-many → a plain association table from sqlalchemy import Column, Table book_tags = Table( "book_tags", Base.metadata, Column("book_id", ForeignKey("books.id", ondelete="CASCADE"), primary_key=True), Column("tag_id", ForeignKey("tags.id", ondelete="CASCADE"), primary_key=True), ) class Book(Base): tags: Mapped[list["Tag"]] = relationship(secondary=book_tags, back_populates="books") ``` - The link row needs extra fields (added_at, role, …)? Drop `secondary` and map the association as a normal model with two FKs — an association object. - `cascade="all, delete-orphan"` makes the ORM delete children when the parent goes or a child is removed from the collection. `ondelete="CASCADE"` is the database doing it. They are independent — use the DB one for correctness under bulk deletes. - `lazy=` sets the DEFAULT loading strategy for a relationship; per-query `.options(...)` overrides it. ## Querying ```python from sqlalchemy import func, select stmt = select(Book).where(Book.title.ilike("%python%")).order_by(Book.title).limit(10) books = session.scalars(stmt).all() # list[Book] book = session.scalars(stmt).first() # Book | None book = session.get(Book, 1) # by PK — hits the identity map first session.scalars(select(Book).where(Book.id == 1)).one() # raises if != 1 row session.scalars(select(Book).where(Book.id == 1)).one_or_none() # raises if > 1 # filters Book.title == "x" # ==, !=, <, > Book.id.in_([1, 2, 3]) Book.bio.is_(None) # never `== None` Book.title.like("a%"), Book.title.ilike("a%") from sqlalchemy import and_, or_, not_ select(Book).where(or_(Book.title == "a", Book.title == "b")) select(Book).where(Book.title == "a", Book.id > 5) # commas = AND # joins & aggregates select(Author).join(Author.books).where(Book.title == "x").distinct() select(Author).outerjoin(Author.books) select(Author.name, func.count(Book.id)).join(Book).group_by(Author.id).having( func.count(Book.id) > 2 ) session.scalar(select(func.count()).select_from(Book)) # a number, not a Book rows = session.execute(select(Book.id, Book.title)).all() # list[Row] of tuples for book_id, title in rows: ... ``` - `scalars()` unwraps the first column — use it when selecting whole entities. `execute()` gives you `Row` tuples, for column selects. - `select()` builds a statement object: reusable, composable, and identical for sync and async. - Filtering on a relationship needs an explicit `join`; `Author.books` alone is a Python attribute, not SQL. ## Eager loading (killing N+1) ```python from sqlalchemy.orm import contains_eager, joinedload, raiseload, selectinload # THE bug: 1 query for the authors, then 1 MORE per author when .books is touched for author in session.scalars(select(Author)): print(author.books) # ← N+1 # collections → selectinload: 2 queries total, a second SELECT ... WHERE id IN (...) session.scalars(select(Author).options(selectinload(Author.books))).all() # many-to-one / one-to-one → joinedload: single query with a LEFT OUTER JOIN session.scalars(select(Book).options(joinedload(Book.author))).all() # nested, one level deeper select(Author).options(selectinload(Author.books).selectinload(Book.tags)) # joinedload on a COLLECTION duplicates parent rows — .unique() is mandatory session.scalars(select(Author).options(joinedload(Author.books))).unique().all() # already joining to filter? reuse that join instead of emitting a second one stmt = ( select(Author) .join(Author.books) .where(Book.title.ilike("%python%")) .options(contains_eager(Author.books)) # NB: only the matching books are loaded ) # make any accidental lazy load blow up — great in tests session.scalars(select(Author).options(raiseload("*"))).all() ``` | Strategy | Queries | Use for | |---|---|---| | `lazy="select"` (default) | 1 per access | nothing — this is the N+1 | | `selectinload` | 2 | collections; safe with `LIMIT` | | `joinedload` | 1 (JOIN) | many-to-one / one-to-one | | `contains_eager` | reuses your join | when you already join to filter | | `subqueryload` | 2 | legacy; `selectinload` is usually better | | `raiseload` | raises | proving a path has no lazy loads | - Rule of thumb: **collection → `selectinload`, scalar → `joinedload`**. - `joinedload` on a collection multiplies parent rows by their children, which both requires `.unique()` and breaks `LIMIT` (the limit counts joined rows). `selectinload` never has that problem. - Set the default on the relationship (`lazy="selectin"`) when a relation is almost always needed, and `lazy="raise"` when it never should be loaded implicitly. - Lazy loading only works while the object is attached to an open session — outside it you get `DetachedInstanceError`. Load what the caller needs before returning. ## Writing data ```python author = Author(name="Ada") session.add(author) session.add_all([book1, book2]) session.flush() # sends INSERT/UPDATE, fills PKs — still inside the transaction session.commit() # flush + COMMIT session.rollback() # discards the transaction AND the in-memory changes session.refresh(author) # re-SELECT this row now author.name = "Ada L." # tracked automatically; UPDATE goes out at the next flush session.delete(author) # ORM delete: runs cascades and events # set-based, no objects loaded: fast, but skips cascades/events/validators from sqlalchemy import delete, update session.execute(update(Book).where(Book.author_id == 1).values(archived=True)) session.execute(delete(Book).where(Book.id.in_(ids))) # upsert (postgres) from sqlalchemy.dialects.postgresql import insert stmt = insert(Tag).values(name="python").on_conflict_do_nothing(index_elements=["name"]) session.execute(stmt) ``` - Autoflush: issuing a query flushes pending changes first, so your own writes are visible to it. That's also why a query can raise an IntegrityError. - `flush()` gives you the generated PK without ending the transaction; `commit()` ends it. Only `commit()` makes anything durable. - Bulk `update()`/`delete()` bypass the ORM — objects already in the session keep their old values. Pass `synchronize_session=False` when you don't care, or re-fetch after. ## Transactions ```python with Session(engine) as session: with session.begin(): # commit at the end, rollback on exception session.add(obj) with session.begin_nested(): # SAVEPOINT — inner failure, outer survives session.add(risky) with engine.begin() as conn: # Core, no ORM/session involved conn.execute(text("UPDATE books SET archived = true")) ``` - A session opens a transaction lazily on first use and holds it until `commit()`/`rollback()`. Long-lived sessions therefore hold long-lived transactions and locks — keep them request-scoped. - After an error you must `rollback()`; the session refuses further work until you do. - `text()` for raw SQL, always with bound parameters: `text("... WHERE id = :id")` + `conn.execute(stmt, {"id": 1})` — never f-strings. ## Alembic ```bash alembic init migrations # -t async for an async setup alembic revision --autogenerate -m "add books table" alembic upgrade head # apply everything pending alembic downgrade -1 # step back one revision alembic current # what the DB thinks it is at alembic history --verbose alembic heads # more than one = branches to merge alembic merge -m "merge heads" heads alembic stamp head # mark as applied WITHOUT running it ``` ```python # migrations/env.py — autogenerate compares this metadata to the live database from myapp.models import Base # importing every model matters: target_metadata = Base.metadata # unimported tables look like drops # a revision file revision = "a1b2c3d4" down_revision = "9f8e7d6c" def upgrade() -> None: op.create_table( "tags", sa.Column("id", sa.Integer(), primary_key=True), sa.Column("name", sa.String(50), nullable=False), ) op.add_column("books", sa.Column("archived", sa.Boolean(), nullable=False, server_default=sa.false())) op.create_index("ix_books_author_id", "books", ["author_id"]) def downgrade() -> None: op.drop_index("ix_books_author_id", "books") op.drop_column("books", "archived") op.drop_table("tags") ``` - Autogenerate is a first draft, not an answer: **read every generated migration.** It detects tables, columns, indexes and type changes, but sees a rename as a drop + add (data loss), and misses most CHECK constraints and server defaults. - Adding a NOT NULL column to a populated table needs either a `server_default`, or three steps: add it nullable → backfill → alter to NOT NULL. - Data migrations: `op.execute("UPDATE ...")`, or `Session(bind=op.get_bind())` for ORM-style work. - SQLite can't `ALTER`: wrap changes in `with op.batch_alter_table("books") as batch_op:` — it rebuilds the table. - Test the round trip (`upgrade head` then `downgrade -1`) before merging, and never edit a migration that has already run somewhere else — write a new one.