ORM & Migrations

What an ORM is, and what it costs honest framing

An ORM maps rows to objects. That is the whole idea. Everything else in this module is the machinery that makes the mapping safe.

What

A library that turns a table row into a Python object and back — Place(id=1, name="Cafe Kranzler") instead of a tuple you index by position.

Where

Between your endpoint code and your database. SQLAlchemy in Python; the same layer exists in every backend stack.

When

Once your schema has more than two tables and you are tired of writing INSERT statements and unpacking cursors by hand.

How

Declare a class per table, hand the class to a session, and the session writes the SQL for you.

In Module 04 you designed six tables and wrote the CREATE TABLE statements yourself. Nothing about that was wasted — it is why you are allowed to use an ORM now. The reason Module 03 (SQL) and Module 04 (schema) came first is that an ORM hides the query. If you had met SQLAlchemy before you met JOIN, the ORM would be magic, and magic that goes slow is impossible to fix.

That last row is the cost, and it is not small. place.tags looks like reading a list. It may be a second round trip to Postgres. Do that inside a loop over fifty places and you have made fifty-one queries where one would have done — the N+1 problem, the single most common performance bug in ORM code anywhere.

This module does not touch your API. Your endpoints still serve the hardcoded Python list you wrote in Module 02, and they will still serve it when this module ends. You are building the database layer beside the app, not inside it. Module 06 does the swap, and it will be a small change precisely because you did this part on its own.

Engine, pool, and session

Three things beginners merge into one blurry idea called "the database connection". They have different lifetimes, and every session bug comes from confusing them.

What

The engine owns a pool of open connections; a session is one unit of work that borrows one connection and tracks the objects you loaded.

Where

One engine per application process, created at import time. One session per request, created and destroyed thousands of times.

When

You create the engine once, ever. You create a session at the start of every operation that touches the database.

How

create_engine(url) once in a module; SessionLocal() per request via the dependency in the next section.

Why a pool exists. Opening a TCP connection to Postgres, authenticating, and setting up the session costs somewhere between a few and a few dozen milliseconds. Doing that per request would dominate the cost of a fast endpoint. So the engine opens a handful of connections up front and hands them out and takes them back — that is all pool_size means: how many are kept open and reused.

# app/db/session.py
import os
from dotenv import load_dotenv
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

load_dotenv()                          # reads .env from Module 00

engine = create_engine(
    os.environ["DATABASE_URL"],        # postgresql://cityguide:...@localhost:5432/cityguide
    echo=False,                        # flip to True when you need to see the SQL
    pool_size=5,                       # connections kept open and reused
    max_overflow=10,                   # extra ones allowed under a burst
    pool_pre_ping=True,                # check a connection is alive before reusing it
)

SessionLocal = sessionmaker(bind=engine, autoflush=False)

You will need the Postgres driver as well as SQLAlchemy itself:

pip install "sqlalchemy>=2.0" psycopg2-binary alembic python-dotenv
pip freeze > requirements.txt

• The session is not the connection. It borrows one when it first needs to talk to the database and gives it back when you close it. A session that never talks to the database never takes a connection at all.
• The session has a memory — the identity map. Load place 7 twice in one session and you get the same Python object both times, from memory, with no second query. This is useful and it is also why stale data inside a long-lived session is a real problem.
• pool_pre_ping=True is worth its tiny cost. Postgres, a container restart, or a network device can close an idle connection without telling you. Without pre-ping your first request after an idle period fails with a dead connection; with it, the pool quietly replaces it.
• autoflush=False is a deliberate choice. With the default, SQLAlchemy secretly sends pending INSERTs before every query so the query sees them. That is convenient and it makes it very hard to predict when your SQL runs. Turning it off means SQL happens when you say flush() or commit(), and nowhere else.

Declaring the six models

This is the Module 04 schema, line for line, expressed as Python classes. Keep your schema.sql open beside it — every constraint you wrote by hand appears here.

What

A model is a Python class whose attributes are columns and whose relationship() attributes are joins waiting to happen.

Where

app/models.py, imported by both your app and by Alembic so it can see the whole schema.

When

After the schema design, never before. The class is a transcription of a decision you already made.

How

SQLAlchemy 2.0 style: subclass DeclarativeBase, annotate with Mapped[...], configure with mapped_column(...).

# app/db/base.py
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    """Every model inherits this. Base.metadata is the catalogue of all tables -
    it is what Alembic reads to work out what your schema should look like."""

# app/models.py
from datetime import datetime

from sqlalchemy import (
    CheckConstraint, Column, DateTime, ForeignKey, String, Table,
    UniqueConstraint, func,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.db.base import Base


# --- place_tags: a pure join table. No extra columns, so it is not a class. ---
place_tags = Table(
    "place_tags",
    Base.metadata,
    Column("place_id", ForeignKey("places.id", ondelete="CASCADE"), primary_key=True),
    Column("tag_id",   ForeignKey("tags.id",   ondelete="CASCADE"), primary_key=True),
)


class Place(Base):
    __tablename__ = "places"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(120))
    neighbourhood: Mapped[str] = mapped_column(String(80), index=True)
    rating: Mapped[int]
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )

    tags: Mapped[list["Tag"]] = relationship(
        secondary=place_tags, back_populates="places"
    )
    list_entries: Mapped[list["ListPlace"]] = relationship(
        back_populates="place", cascade="all, delete-orphan"
    )

    __table_args__ = (
        CheckConstraint("rating BETWEEN 1 AND 5", name="ck_places_rating_range"),
        UniqueConstraint("name", "neighbourhood", name="uq_places_name_hood"),
    )


class Tag(Base):
    __tablename__ = "tags"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(40), unique=True)

    places: Mapped[list["Place"]] = relationship(
        secondary=place_tags, back_populates="tags"
    )


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)
    display_name: Mapped[str] = mapped_column(String(80))
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )

    lists: Mapped[list["SavedList"]] = relationship(
        back_populates="owner", cascade="all, delete-orphan"
    )


class SavedList(Base):
    __tablename__ = "lists"          # class name and table name need not match

    id: Mapped[int] = mapped_column(primary_key=True)
    owner_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"))
    name: Mapped[str] = mapped_column(String(80))
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )

    owner: Mapped["User"] = relationship(back_populates="lists")
    entries: Mapped[list["ListPlace"]] = relationship(
        back_populates="saved_list",
        cascade="all, delete-orphan",
        order_by="ListPlace.position",
    )

    __table_args__ = (
        UniqueConstraint("owner_id", "name", name="uq_lists_owner_name"),
    )


# --- list_places: a join table WITH extra columns, so it IS a class. ---
class ListPlace(Base):
    __tablename__ = "list_places"

    list_id: Mapped[int] = mapped_column(
        ForeignKey("lists.id", ondelete="CASCADE"), primary_key=True
    )
    place_id: Mapped[int] = mapped_column(
        ForeignKey("places.id", ondelete="CASCADE"), primary_key=True
    )
    position: Mapped[int] = mapped_column(default=0)
    added_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )

    saved_list: Mapped["SavedList"] = relationship(back_populates="entries")
    place: Mapped["Place"] = relationship(back_populates="list_entries")

Line by line, against the DDL you wrote in Module 04. Mapped[int] = mapped_column(primary_key=True) is your id SERIAL PRIMARY KEY. Mapped[str] with no | None is NOT NULL — SQLAlchemy 2.0 reads the type annotation, so nullability comes from the type, not from a keyword. Mapped[str | None] is a nullable column. CheckConstraint("rating BETWEEN 1 AND 5") is the exact CHECK you typed. server_default=func.now() is DEFAULT now() — note server default: Postgres fills it in, not Python, so a row inserted by psql gets a timestamp too.

Why two different shapes for two many-to-manys. place_tags only says "this place has this tag" — nothing else belongs on the row, so it is a bare Table and you never see it: you work with place.tags. list_places carries position and added_at, because the order of places in a saved list is real information. A row with its own data is an entity, so it gets a class. This choice will come up in every schema you ever build.

back_populates is what keeps the two sides honest. It tells SQLAlchemy that Place.tags and Tag.places are two views of the same join table, so appending to one updates the other in memory without a reload. Omit it and you get two independent relationships that silently disagree.

Querying is select(), not the old .query():

from sqlalchemy import select
from sqlalchemy.orm import selectinload

# SELECT * FROM places WHERE neighbourhood = 'Kreuzberg' ORDER BY rating DESC LIMIT 5
stmt = (
    select(Place)
    .where(Place.neighbourhood == "Kreuzberg")
    .order_by(Place.rating.desc())
    .limit(5)
)
places = db.scalars(stmt).all()        # -> list[Place]

one = db.get(Place, 7)                 # fetch by primary key; None if missing
maybe = db.scalars(select(Place).where(Place.name == "x")).one_or_none()

# Load the tags in ONE extra query instead of one per place - kills the N+1
rows = db.scalars(select(Place).options(selectinload(Place.tags)).limit(50)).all()

The session lifecycle read twice

This is the most important section in the module and the source of the three bugs that will actually cost you an evening. A session is a short-lived scratchpad, not a database handle you keep around.

What

A session is one unit of work: it opens a transaction, tracks every object you load or add, and ends when you close it.

Where

Created per request by a FastAPI dependency, handed to your endpoint as an argument, closed automatically when the response is sent.

When

Every request that touches the database gets its own. Never a module-level db = SessionLocal().

How

A generator dependency — the get_db function below — which you will import in Module 06 and every module after.

# app/db/session.py  (continued)
from collections.abc import Generator
from sqlalchemy.orm import Session

def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db          # <- everything before this runs BEFORE your endpoint
    finally:
        db.close()        # <- this runs AFTER the response is finished

What yield means in a dependency. FastAPI calls get_db(), runs it up to the yield, and passes the yielded value into your endpoint as the parameter. Your endpoint runs. When the response has been sent, FastAPI resumes the generator from where it paused and runs the rest — here, the finally block. So yield splits the function into setup and teardown around the request, and the try/finally guarantees the teardown happens even when your endpoint raises.

You use it like this — and this is exactly the signature every endpoint gets in Module 06:

from fastapi import Depends
from sqlalchemy.orm import Session
from app.db.session import get_db

@app.get("/places")
def list_places(db: Session = Depends(get_db)):
    return db.scalars(select(Place).limit(20)).all()

The three classic failures. Each has a name, a symptom, and a fix.

(a) A shared session leaks one request's data into another. Write db = SessionLocal() at the top of a module and every request in that process uses the same session. Its identity map now holds objects from other people's requests, uncommitted changes from request A are visible to request B, and a failed rollback in one request poisons the next. Under a single-user test everything looks fine. Under two concurrent users you get data belonging to a stranger, which is a security incident, not a bug.

(b) DetachedInstanceError from touching a lazy attribute after the session closed. A Place you loaded is bound to the session that loaded it. When the session closes, the object survives but its link to the database does not — it is detached. Any attribute that has not been loaded yet, especially a relationship(), now has no way to fetch itself:

sqlalchemy.orm.exc.DetachedInstanceError: Parent instance <Place at 0x104a2f10>
is not bound to a Session; lazy load operation of attribute 'tags' cannot proceed

This is not a mysterious framework failure. It is SQLAlchemy telling you: you asked me for data I would need a database to fetch, and you took the database away. The two real fixes are to load what you need while the session is open (selectinload), or to convert the object into a plain response model before the session closes — which is exactly what the Pydantic response schemas in Module 06 do, and one reason they exist.

(c) Forgetting to close, and exhausting the pool. Every session that is never closed keeps its borrowed connection forever. With pool_size=5, max_overflow=10 your app serves fifteen requests happily and then stops — not crashes, stops, with every new request hanging until it times out:

sqlalchemy.exc.TimeoutError: QueuePool limit of size 5 overflow 10 reached,
connection timed out, timeout 30.00

You will build this failure deliberately in the project below, because the symptom — "the app worked all morning and now every request hangs" — is one you want to recognise instantly rather than debug from first principles at 6pm.

• Short-lived is the whole discipline. Open late, close early. A session held open across a slow HTTP call to another service is holding a database connection hostage for no reason.
• try/finally, not try/except, for closing. finally runs whether the endpoint returned or raised. An unclosed session after an exception is exactly the case that exhausts your pool, because errors cluster.
• Outside a request, use the context manager. Scripts and background jobs have no dependency injection: with SessionLocal() as db: gives the same guarantee in one line.
• Never return a model object from a function whose session has closed and expect its relationships to work. Return what you actually need, loaded, or a dict.

Transactions, flush, and rollback

A transaction is a group of writes that either all happen or none of them do. A partial write is worse than a failed write, because a failure is loud and a half-written record is silent.

What

An all-or-nothing block of database work. Postgres holds your changes invisible to everyone else until you commit.

Where

A session opens one the first time it talks to the database and ends it at commit() or rollback().

When

Any operation with more than one write. Creating a list and adding places to it is two writes and therefore one transaction.

How

db.flush() to send SQL and keep going; db.commit() to make it permanent; db.rollback() to throw it all away.

Flush exists because of ids. You need the new list's id to insert its list_places rows, and only Postgres can generate that id. flush() sends the INSERT, reads back the id, and leaves the transaction open — so if step two fails, step one is still undoable. That is the entire reason the two operations are separate.

# app/repository.py
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models import Place, SavedList, ListPlace


def create_list_with_places(
    db: Session, owner_id: int, name: str, place_ids: list[int]
) -> SavedList:
    saved = SavedList(owner_id=owner_id, name=name)
    db.add(saved)
    db.flush()            # INSERT INTO lists ... RETURNING id
                          # saved.id is now real. Transaction still OPEN.

    for position, place_id in enumerate(place_ids):
        place = db.get(Place, place_id)
        if place is None:
            # Step two fails. The list from step one must not survive.
            raise ValueError(f"place {place_id} does not exist")
        db.add(ListPlace(list_id=saved.id, place_id=place_id, position=position))

    db.commit()           # both writes become permanent, together
    db.refresh(saved)     # re-read the row so server defaults are populated
    return saved

# the caller - this shape goes in your endpoint in Module 06
try:
    saved = create_list_with_places(db, owner_id=1, name="Sunday walk",
                                    place_ids=[1, 2, 9999])   # 9999 does not exist
except Exception:
    db.rollback()         # the lists row from the flush is discarded
    raise

Run that and then count the rows in lists. Zero new rows. The flush() genuinely sent an INSERT to Postgres — you could have printed saved.id and seen a number — and the rollback() unwound it. Without the rollback you would have an empty saved list in your database that no user ever asked for and no code will ever clean up.

def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db
    except Exception:
        db.rollback()
        raise             # re-raise so FastAPI still returns an error response
    finally:
        db.close()

• Commit once, at the end of the operation — not after each add. Committing in a loop turns one atomic operation into N independent ones, and the failure of number four leaves three permanent rows behind.
• Do not commit inside a repository function that a bigger operation might call. The caller owns the transaction boundary, because only the caller knows where the operation ends. This is a real design rule, not style.
• After a failed statement the session is unusable until you roll back. Postgres aborts the transaction; further queries return "current transaction is aborted, commands ignored until end of transaction block". That message means "you missed a rollback", nothing more exotic.
• refresh() after commit re-reads the row. Server defaults like created_at were computed by Postgres, so your Python object does not know them until you ask.

Alembic: schema change you can repeat

You changed a table by typing ALTER TABLE into psql. It worked. Now do it again on the machine of someone who is not you, in the right order, six months from now.

What

Alembic records each schema change as a numbered Python file in your repository, so the database's shape is version-controlled alongside the code that depends on it.

Where

A migrations/ folder in your project, committed to git, plus one tiny alembic_version table inside the database.

When

Every single schema change from now on. Once Alembic is set up, manual ALTER TABLE is a bug.

How

Change the model, autogenerate a migration, read it, then alembic upgrade head.

The problem it solves. Manual DDL leaves no record of what was run or in what order. Your machine has the column; the test database does not; the second developer's does but with a different type. The code assumes one shape and the database has another, and the error appears at runtime, in production, on a Friday. A migration file makes the change a reviewable artefact that anyone can replay from empty to current with one command.

alembic init migrations        # creates migrations/ and alembic.ini

Two edits before it will work. In alembic.ini, comment out the hardcoded URL — your real one lives in .env and must never be committed:

# alembic.ini
# sqlalchemy.url = driver://user:pass@localhost/dbname     <- comment this out

# migrations/env.py  - add near the top, after "config = context.config"
import os
from dotenv import load_dotenv

from app.db.base import Base
import app.models          # noqa: F401  - side-effect import: registers every
                           # model on Base.metadata. WITHOUT THIS LINE,
                           # autogenerate sees no tables and drops your schema.

load_dotenv()
config.set_main_option("sqlalchemy.url", os.environ["DATABASE_URL"].replace("%", "%%"))

target_metadata = Base.metadata      # replaces "target_metadata = None"

The loop, in the order you will run it forever.

alembic revision --autogenerate -m "initial schema"   # 1. write a draft
                                                       # 2. READ THE FILE
alembic upgrade head                                   # 3. apply it
alembic current                                        # 4. which revision am I on?
alembic history --verbose                              # 5. the whole chain
alembic downgrade -1                                   # 6. undo the last one

A generated migration looks like this. The two long hex strings are the point — each file names itself and its parent, which is how Alembic knows the order:

# migrations/versions/a1b2c3d4e5f6_initial_schema.py
revision = "a1b2c3d4e5f6"
down_revision = None            # this one is first; the next will name this id

def upgrade() -> None:
    op.create_table(
        "places",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("name", sa.String(length=120), nullable=False),
        ...
        sa.CheckConstraint("rating BETWEEN 1 AND 5", name="ck_places_rating_range"),
        sa.PrimaryKeyConstraint("id"),
    )
    ...

def downgrade() -> None:
    op.drop_table("places")
    ...

What alembic upgrade head actually does. It reads the alembic_version table — one column, one row, holding the id of the last migration applied to this database. It builds the chain of revisions from your migrations/versions/ folder by following each file's down_revision. It then runs the upgrade() function of every revision after your current one, in order, one at a time, updating alembic_version after each. "head" means the last revision in the chain. On an empty database, alembic_version does not exist, so Alembic creates it and runs everything from the start.

Migrations are not seeds. A migration changes the shape of the database. A seed script puts data in it. Keep them apart: seeds are for your laptop and your tests, they change constantly, and putting them in a migration means every future environment permanently replays your fake cafes. Seeding is an ordinary script that uses the ORM you just built — and it must be idempotent, because you will run it more than once:

# seed.py
from sqlalchemy import select
from app.db.session import SessionLocal
from app.models import Place, Tag

PLACES = [
    ("Cafe Kranzler",  "Charlottenburg", 4, ["coffee", "cake"]),
    ("Markthalle Neun","Kreuzberg",      5, ["food", "market"]),
    ("Tempelhofer Feld","Tempelhof",     5, ["outdoors", "free"]),
]

def get_or_create_tag(db, name: str) -> Tag:
    tag = db.scalars(select(Tag).where(Tag.name == name)).one_or_none()
    if tag is None:
        tag = Tag(name=name)
        db.add(tag)
        db.flush()                     # need its id before linking
    return tag

def main() -> None:
    with SessionLocal() as db:         # context manager closes it for us
        for name, hood, rating, tag_names in PLACES:
            existing = db.scalars(
                select(Place).where(Place.name == name,
                                    Place.neighbourhood == hood)
            ).one_or_none()
            if existing is not None:
                continue               # idempotent: already seeded, skip
            place = Place(name=name, neighbourhood=hood, rating=rating)
            place.tags = [get_or_create_tag(db, t) for t in tag_names]
            db.add(place)
        db.commit()
        print("seeded")

if __name__ == "__main__":
    main()

Notice place.tags = [...]. You never touch the place_tags table — you assign a list of Tag objects and SQLAlchemy writes the join rows. That is the ORM earning its keep, and it is also why the UniqueConstraint from Module 04 still matters: the ORM will happily insert a duplicate place if you let it, and the database is the only thing that will not.

Build this

Verify it yourself

Run every command with your venv active, from the project root. Any mismatch means the module is not finished — which is information, not failure.

$ alembic current
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
9c1a4f2b7d10 (head)                  # your ids differ; "(head)" must be there

$ alembic history
a1b2c3d4e5f6 -> 9c1a4f2b7d10 (head), add places.website
<base> -> a1b2c3d4e5f6, initial schema
                                     # exactly two revisions, newest first

$ docker compose exec db psql -U cityguide -d cityguide -c "SELECT version_num FROM alembic_version;"
  version_num
--------------
 9c1a4f2b7d10                        # matches "alembic current" above
(1 row)

$ docker compose exec db psql -U cityguide -d cityguide -c "\d places"
# read the output: 6 columns, and at the bottom you must see
#   Check constraints: "ck_places_rating_range" CHECK (rating >= 1 AND rating <= 5)
#   Indexes: "uq_places_name_hood" UNIQUE CONSTRAINT, btree (name, neighbourhood)

$ docker compose exec db psql -U cityguide -d cityguide -c "SELECT column_name, data_type, is_nullable FROM information_schema.columns WHERE table_name='places' ORDER BY ordinal_position;"
  column_name  |          data_type          | is_nullable
---------------+-----------------------------+-------------
 id            | integer                     | NO
 name          | character varying           | NO
 neighbourhood | character varying           | NO
 rating        | integer                     | NO
 created_at    | timestamp with time zone    | NO
 website       | character varying           | YES        # <- the second migration
(6 rows)

$ python seed.py && python seed.py
seeded
seeded                               # ran twice, on purpose

$ docker compose exec db psql -U cityguide -d cityguide -c "SELECT (SELECT count(*) FROM places) AS places, (SELECT count(*) FROM tags) AS tags, (SELECT count(*) FROM place_tags) AS links;"
 places | tags | links
--------+------+-------
      3 |    6 |     6               # NOT 6 places - the seed is idempotent
(1 row)

$ python - <<'PY'
from sqlalchemy import select
from app.db.session import engine, SessionLocal
from app.models import Place
engine.echo = True                   # print every statement SQLAlchemy sends
with SessionLocal() as db:
    print([p.name for p in db.scalars(select(Place).order_by(Place.id))])
PY
INFO sqlalchemy.engine.Engine SELECT places.id, places.name, places.neighbourhood,
places.rating, places.created_at, places.website FROM places ORDER BY places.id
INFO sqlalchemy.engine.Engine [generated in 0.00012s] {}
['Cafe Kranzler', 'Markthalle Neun', 'Tempelhofer Feld']
                                     # real SQL, printed. This is the tool you
                                     # reach for whenever an endpoint is slow.

Prove it

Write your answer to each in notes/module-05.md in your own words first, then open the answer. Where you disagree, work out which of you is right — that gap is the actual learning.

1. Step by step, what does alembic upgrade head do to the database?

2. Why must a session be created per request rather than once at module level?

3. What is the difference between flush and commit, and when do you genuinely need flush?

4. DetachedInstanceError — what is it actually telling you, and what are the two real fixes?

5. Why must you read an autogenerated migration before running it? Give two concrete things it gets wrong.

6. place_tags is a plain Table but list_places is a class. Why the difference?