Week 5: Relationships, Transactions & Alembic

1. Relationships & Foreign Keys

A one-to-many relationship — one project, many tasks — is expressed with a foreign key column on the "many" side plus a relationship() on both model classes. The foreign key is what actually exists in the database; relationship() is a Python-only convenience that lets you navigate the connection as attributes (project.tasks, task.project) instead of writing the join yourself every time:

from sqlalchemy import ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.db.base import Base


class Project(Base):
    __tablename__ = "projects"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    tasks: Mapped[list["Task"]] = relationship(back_populates="project")


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    project_id: Mapped[int | None] = mapped_column(ForeignKey("projects.id"))
    project: Mapped["Project | None"] = relationship(back_populates="tasks")

back_populates links the two relationship() calls together as two ends of the same association: assign task.project = some_project and SQLAlchemy automatically appends task to some_project.tasks in memory too, keeping both sides consistent without a second query. Skip back_populates on one side and the two attributes become independent — the foreign key column in the database is still correct, but your in-memory object graph can silently drift out of sync with it.

Many-to-many relationships — tasks that can each carry several tags, and tags reused across many tasks — need a third table in between, an association table, since a plain foreign key column can only point to one row:

from sqlalchemy import Column, ForeignKey, Table

from app.db.base import Base

task_tags = Table(
    "task_tags",
    Base.metadata,
    Column("task_id", ForeignKey("tasks.id"), primary_key=True),
    Column("tag_id", ForeignKey("tags.id"), primary_key=True),
)


class Tag(Base):
    __tablename__ = "tags"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(unique=True)
    tasks: Mapped[list["Task"]] = relationship(secondary=task_tags, back_populates="tags")

secondary=task_tags tells SQLAlchemy to route the relationship through that join table automatically — appending to task.tags inserts a row into task_tags behind the scenes, and removing an entry deletes it, without you ever writing raw SQL against the association table yourself.

2. Loading Strategies & the N+1 Problem

By default, accessing a relationship() attribute that hasn't been loaded yet triggers lazy loading: SQLAlchemy quietly runs a new query the moment you touch project.tasks. That's convenient in isolation, but it becomes a real performance problem the moment you loop over multiple parent rows — this is the classic N+1 problem.

Concretely: select(Project) for 20 projects runs 1 query. Then, if your serializer or route handler reads project.tasks for each of those 20 projects, lazy loading fires one additional query per project — 20 more queries, each fetching one project's tasks separately. That's 21 total round trips to render one response, and the number keeps growing linearly with however many projects you return. Eager loading with selectinload() collapses that back down to a fixed, bounded number:

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

from app.models.project import Project


def list_projects_with_tasks(session: Session) -> list[Project]:
    stmt = select(Project).options(selectinload(Project.tasks))
    return list(session.scalars(stmt))

selectinload(Project.tasks) runs exactly 2 queries total regardless of how many projects come back: one SELECT for the projects, and one follow-up SELECT ... WHERE task.project_id IN (...) that pulls every related task for every returned project in a single round trip, using the already-known project ids. SQLAlchemy then matches each task back to its project in Python. Twenty projects went from 21 queries down to 2 — and the count stays at 2 even at two hundred projects, which is the actual fix, not just a smaller number.

3. Atomic Transactions & Alembic Migrations

Any operation that writes to more than one row — or more than one table — should commit or fail as a single unit. session.begin() used as a context manager gives you exactly that: it commits automatically if the block finishes cleanly, and rolls back automatically if anything inside it raises, before the exception propagates:

def bulk_add_tasks(session: Session, project: Project, titles: list[str]) -> list[Task]:
    created: list[Task] = []
    with session.begin():
        for title in titles:
            task = Task(title=title, project_id=project.id)
            session.add(task)
            created.append(task)
    # Exiting the "with" block cleanly triggers commit().
    # An exception raised inside it triggers rollback() instead --
    # either every task above is persisted, or none of them are.
    return created

Without that boundary, a failure on the third of five inserts would leave the first two committed and the database in a state your application never intended to produce. Wrapping the whole operation in one transaction means a partial failure leaves the database exactly as it was before the request started — there's no in-between state to reason about.

Now that your models actually have relationships, changing them by hand-editing the database is too error-prone to trust — that's what Alembic is for. It compares your SQLAlchemy models against the live database schema and generates a versioned, reviewable migration script:

# One-time setup
poetry add alembic
poetry run alembic init alembic

# In alembic/env.py, point target_metadata at your models:
#   from app.db.base import Base
#   target_metadata = Base.metadata

# Autogenerate a migration from the current model definitions
poetry run alembic revision --autogenerate -m "add projects and task ownership"

# ALWAYS read the generated script in alembic/versions/ before applying it

# Apply it to the database
poetry run alembic upgrade head

Autogenerate is a heuristic, not a guarantee — it detects new tables and new columns reliably, but it can misread a renamed column as an unrelated "drop old column, add new column" pair, which silently discards existing data on upgrade. Reading the generated script in alembic/versions/ before running upgrade head is a review step, not an optional formality.

4. Hands-on Exercise

5. Knowledge Check

Four quick questions. Expand each to check your answer.