Module 3 · Postgres & Data · Deep Dive

SQLAlchemy 2.0 & Migrations

Stop writing raw SQL strings by hand. Map Python classes to Postgres tables, query in pure Python, and version-control your schema with Alembic — the way every modern Python backend ships.

BasicIntermediateBuild

1 · What an ORM is

An ORM (Object-Relational Mapper) is a translation layer. A Python class maps to a database table; an instance of that class maps to a row; an attribute maps to a column. You work with normal Python objects, and the ORM generates the SQL to load and save them.

Why bother instead of raw SQL? Three reasons interviewers want to hear: you stay in one language (no string-glued SQL scattered everywhere), you get safety (parameters are bound, so SQL injection is much harder), and you get portability and refactoring (rename a column in one model, not across fifty query strings). The trade-off: a layer of abstraction you must understand to debug.

2 · 2.0 declarative models

SQLAlchemy 2.0 is a real break from the old 1.x tutorials you'll find online. The modern way uses a DeclarativeBase subclass plus typed attributes with Mapped[...] and mapped_column(). The type annotation drives the column type, so models read like ordinary typed Python.

models.py

from datetime import datetime
from sqlalchemy import String, ForeignKey, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)
    name: Mapped[str | None] = mapped_column(String(100))


class Document(Base):
    __tablename__ = "documents"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    body: Mapped[str]
    owner_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

Read the annotations carefully — they carry meaning. Mapped[int] with primary_key=True is an auto-incrementing PK. Mapped[str] is NOT NULL; Mapped[str | None] allows NULL. A bare Mapped[str] with no mapped_column still becomes a column — the annotation alone is enough.

3 · Engine & Session

Two objects connect your models to the database. The Engine is the connection factory — created once, at startup, for the whole app. The Session is your unit of work — short-lived, one per request, the thing you actually add and query through.

db.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# postgresql+psycopg = the Postgres dialect using the psycopg 3 driver
engine = create_engine("postgresql+psycopg://user:pass@localhost:5432/docchat")

# A factory that produces new Session objects
SessionLocal = sessionmaker(bind=engine)

The connection string has a fixed shape: dialect+driver://user:pass@host:port/database. In real code the URL comes from an environment variable, never a hardcoded literal.

A session's lifecycle is four moves. You add objects to it, commit to flush them to Postgres in a transaction, optionally refresh to pull back server-generated values (like the new id), and close to release the connection.

with SessionLocal() as session:
    user = User(email="sam@uae.dev", name="Sam")
    session.add(user)        # staged, not yet in the DB
    session.commit()         # transaction committed
    session.refresh(user)    # now user.id is populated
    print(user.id)
# leaving the `with` block closes the session automatically

4 · CRUD through the ORM

This is the day-to-day. In 2.0 you read with select() and execute it through the session — the old session.query(...) API still exists but is legacy. Get used to select().

Create add & commit

doc = Document(title="Q2 Report", body="...", owner_id=1)
session.add(doc)
session.commit()

Read select / where / order_by

from sqlalchemy import select

stmt = (
    select(Document)
    .where(Document.owner_id == 1)
    .order_by(Document.created_at.desc())
)
docs = session.scalars(stmt).all()   # list[Document], newest first

# A single row (or None):
one = session.scalars(
    select(Document).where(Document.id == 5)
).first()

Note session.scalars(stmt).all() — scalars() unwraps each row to the single ORM object, so you get a clean list[Document] rather than rows of tuples. Use .first() for one-or-none.

Update mutate & commit

doc = session.scalars(
    select(Document).where(Document.id == 5)
).first()
doc.title = "Q2 Report (final)"   # just set the attribute
session.commit()                     # SQLAlchemy issues the UPDATE

You don't write an UPDATE statement — you change the object and commit. The session tracks what changed and emits the SQL. This "dirty tracking" is core ORM behaviour.

Delete remove & commit

session.delete(doc)
session.commit()

5 · Relationships & loading

A ForeignKey gives you the column; a relationship() gives you the convenient Python attribute that navigates it. One User owns many Documents — a classic one-to-many. back_populates wires both sides together so they stay in sync.

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)

    documents: Mapped[list["Document"]] = relationship(
        back_populates="owner"
    )


class Document(Base):
    __tablename__ = "documents"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    owner_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    owner: Mapped["User"] = relationship(back_populates="documents")

Now navigation is just attribute access:

user.documents      # list[Document] owned by this user
doc.owner.email     # walk from a document back to its user

6 · Wiring into FastAPI

Back in Module 2 you learned dependency injection with Depends. The session fits that pattern perfectly: a get_db dependency that yields a session and guarantees it closes after the request — even if the route raises.

deps.py

from collections.abc import Generator
from sqlalchemy.orm import Session
from .db import SessionLocal


def get_db() -> Generator[Session]:
    db = SessionLocal()
    try:
        yield db          # hand the session to the route
    finally:
        db.close()       # always runs, even on error

main.py

from fastapi import Depends, FastAPI
from sqlalchemy import select
from sqlalchemy.orm import Session
from .deps import get_db
from .models import Document

app = FastAPI()


@app.get("/documents")
def list_documents(db: Session = Depends(get_db)):
    docs = db.scalars(
        select(Document).order_by(Document.created_at.desc())
    ).all()
    return docs

The yield is the whole trick: code before it runs at request start, code in finally runs at request end. One dependency, reused by every route that touches the database.

7 · Alembic migrations

Your models describe the schema you want. The database has the schema it currently has. Alembic (SQLAlchemy's migration tool) bridges the gap and — crucially — version-controls it, so every change is a reviewable, repeatable file committed to git. No more "did anyone run the ALTER on prod?"

# 1. One-time setup — creates the alembic/ folder + config
alembic init alembic

# 2. Compare models to the DB and auto-generate a revision
alembic revision --autogenerate -m "create users and documents"

# 3. Apply all pending revisions to the database
alembic upgrade head

# (to undo the latest revision)
alembic downgrade -1

To make --autogenerate work, point Alembic at your models' metadata. In alembic/env.py set target_metadata = Base.metadata so Alembic can diff your classes against the live schema and write the op.create_table(...) / op.add_column(...) calls for you.

8 · Build it

Before: documents lived in a module-level list that reset on every reload.

before — in-memory (don't keep this)

DOCS = []   # lost on restart

@app.post("/documents")
def create(title: str, body: str):
    DOCS.append({"title": title, "body": body})
    return {"count": len(DOCS)}

After: the same endpoints, now writing to and reading from Postgres through the session.

after — persisted in Postgres

from fastapi import Depends, FastAPI
from sqlalchemy import select
from sqlalchemy.orm import Session
from .deps import get_db
from .models import Document

app = FastAPI()


@app.post("/documents")
def create(title: str, body: str, db: Session = Depends(get_db)):
    doc = Document(title=title, body=body, owner_id=1)
    db.add(doc)
    db.commit()
    db.refresh(doc)            # pull back the new id
    return doc


@app.get("/documents")
def list_docs(db: Session = Depends(get_db)):
    return db.scalars(
        select(Document).order_by(Document.created_at.desc())
    ).all()

Steps in order: define Base, User, Document in models.py; create the engine and SessionLocal in db.py; add get_db; then alembic revision --autogenerate and alembic upgrade head to create the tables. Restart, POST a document, restart again — it's still there.

9 · Many-to-many relationships

One-to-many covered User → Document. But a Document can carry many tags, and each tag labels many documents — that's many-to-many. SQL has no native column for this, so you model it with a third association table holding just the two foreign keys. In 2.0 you declare a plain Table and hand it to relationship(secondary=...).

models.py

from sqlalchemy import Table, Column, ForeignKey, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


# Association table: just the two foreign keys, no model class
document_tag = Table(
    "document_tag",
    Base.metadata,
    Column("document_id", ForeignKey("documents.id"), primary_key=True),
    Column("tag_id", ForeignKey("tags.id"), primary_key=True),
)


class Document(Base):
    __tablename__ = "documents"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))

    tags: Mapped[list["Tag"]] = relationship(
        secondary=document_tag, back_populates="documents"
    )


class Tag(Base):
    __tablename__ = "tags"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True)

    documents: Mapped[list["Document"]] = relationship(
        secondary=document_tag, back_populates="tags"
    )

Now linking is pure Python — append a Tag object to doc.tags and commit; SQLAlchemy inserts the join row for you:

doc.tags.append(Tag(name="finance"))
session.commit()        # a row appears in document_tag

10 · Cascades & delete behavior

In DocChat a Document gets split into chunks for embedding. Those chunks are meaningless without their parent — delete the document and the chunks must die with it. By default SQLAlchemy does not do that; you ask for it with cascade="all, delete-orphan" on the parent side.

class Document(Base):
    __tablename__ = "documents"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))

    chunks: Mapped[list["Chunk"]] = relationship(
        back_populates="document",
        cascade="all, delete-orphan",
        passive_deletes=True,
    )


class Chunk(Base):
    __tablename__ = "chunks"
    id: Mapped[int] = mapped_column(primary_key=True)
    text: Mapped[str]
    document_id: Mapped[int] = mapped_column(
        ForeignKey("documents.id", ondelete="CASCADE")
    )

    document: Mapped["Document"] = relationship(back_populates="chunks")

Two distinct things are happening, and interviewers probe the difference. cascade="all, delete-orphan" is the ORM-level rule: deleting a Document deletes its Chunks, and removing a chunk from doc.chunks (orphaning it) deletes that chunk too. The ondelete="CASCADE" on the ForeignKey is the database-level rule — Postgres itself removes the rows. passive_deletes=True tells SQLAlchemy to trust the database to do it, so it skips loading every child into memory just to delete them one by one.

doc = session.get(Document, 5)
session.delete(doc)
session.commit()        # the document AND all its chunks are gone

11 · Loading strategies compared

Section 5 met the N+1 trap and fixed it with selectinload. But that's one of several strategies, and picking the right one per relationship is a senior skill. You pass them through .options(...) on a select().

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

# Collections (one-to-many / many-to-many): selectinload
stmt = select(User).options(selectinload(User.documents))

# Many-to-one (the single parent): joinedload
stmt = select(Document).options(joinedload(Document.owner))

The rule of thumb: collections want selectinload (a single IN query, no row multiplication), while a many-to-one wants joinedload (one JOIN, no extra round trip). Avoid joinedload on a collection — the join multiplies the parent row once per child and bloats the result set. Set lazy="raise" on a relationship during development so any forgotten eager-load blows up loudly instead of silently firing N queries in prod.

12 · Connection pooling & engine lifecycle

Opening a Postgres connection is expensive — a TCP handshake plus auth, every time. The Engine avoids that by keeping a pool of live connections and lending one out per session, then taking it back on close() rather than tearing it down. You create the engine once for the whole process and tune the pool with arguments.

db.py

import os
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine(
    os.environ["DATABASE_URL"],
    pool_size=5,            # persistent connections kept open
    max_overflow=10,        # extra burst connections under load
    pool_pre_ping=True,     # test a connection before handing it out
)

SessionLocal = sessionmaker(bind=engine)

pool_size is the steady-state pool; max_overflow is how many extra connections it may open during a spike before requests have to wait. The one that saves you in production is pool_pre_ping=True: it sends a tiny test query before lending a connection, and silently reconnects if it's dead.

The engine lives for the whole app; the session lives for one request. That's exactly the get_db dependency from section 6 — SessionLocal() borrows a connection from the pool, db.close() in the finally returns it. One session per request, one engine per process: the standard FastAPI shape.

13 · Hybrid properties

Sometimes you want a computed attribute that works the same in Python and in SQL — readable as obj.prop on a loaded object, and usable inside a where() so the database does the filtering. That's @hybrid_property.

from sqlalchemy.ext.hybrid import hybrid_property


class Document(Base):
    __tablename__ = "documents"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    body: Mapped[str]

    @hybrid_property
    def is_long(self) -> bool:
        return func.length(self.body) > 5000

The same expression evaluates in Python on an instance and compiles to SQL in a query:

doc.is_long                         # True/False on a loaded object
session.scalars(
    select(Document).where(Document.is_long)   # filters in Postgres
).all()

14 · Check yourself

Answer from memory — retrieval is what moves this from "I read it" to "I know it".