Week 4: SQLAlchemy 2.0 Fundamentals

1. Declarative Models & Mapped Columns

SQLAlchemy 2.0 introduced a fully typed declarative style built around DeclarativeBase, Mapped[] and mapped_column(). You still get the same ORM underneath, but every column is now expressed as a regular Python type annotation that your editor and mypy understand — no more guessing whether a column returns str or str | None from a bare Column(String) declaration in the legacy 1.x style.

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    """Shared base class -- every mapped model inherits from this."""
    pass

Every model class inherits from that one Base, and each attribute is declared as Mapped[SomeType]. The Mapped[] wrapper is what tells SQLAlchemy's mapper "this attribute is backed by a database column" and gives type checkers the real Python type; mapped_column() is where you configure the column-level details — primary key, server default, nullability:

from datetime import datetime

from sqlalchemy import func
from sqlalchemy.orm import Mapped, mapped_column

from app.db.base import Base


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    description: Mapped[str | None] = mapped_column(default=None)
    is_done: Mapped[bool] = mapped_column(default=False)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

Nullability is inferred from the type itself: title: Mapped[str] becomes a NOT NULL column because str doesn't include None, while description: Mapped[str | None] becomes nullable automatically — you rarely need to pass nullable=True explicitly. server_default=func.now() asks Postgres itself to fill in the timestamp at insert time, which is more reliable than computing it in Python, since it stays correct even if a row is inserted through a raw SQL script that bypasses your application entirely.

2. Session Lifecycle as a FastAPI Dependency

A Session is the ORM's unit of work: it tracks every object you load or add, translates your query calls into SQL over a connection it borrows from a pool, and stages changes until you explicitly commit(). Creating one starts with an engine — the object that actually knows how to talk to Postgres and manages the underlying connection pool — built once at startup from a connection URL:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.core.config import settings

engine = create_engine(settings.database_url, pool_pre_ping=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)


def get_session():
    session = SessionLocal()
    try:
        yield session
    finally:
        session.close()

get_session() is a generator dependency: FastAPI calls it, receives the value produced by yield, injects that Session into your route, and — critically — resumes the generator after the response has been sent, running whatever comes after yield as guaranteed cleanup. That's what makes session.close() run for every request, success or failure, without you having to remember it in each route.

from typing import Annotated

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session

from app.db.session import get_session

router = APIRouter(prefix="/tasks", tags=["tasks"])
SessionDep = Annotated[Session, Depends(get_session)]


@router.get("/{task_id}")
def get_task(task_id: int, session: SessionDep):
    ...

A single module-level Session shared across requests would be a serious bug, not a shortcut: Session objects are not thread-safe, so two concurrent requests touching the same instance can corrupt each other's identity map and issue queries on top of each other's uncommitted changes. Creating a fresh Session per request — and closing it when the request ends — is what keeps each request's work isolated and returns its connection to the pool promptly instead of holding it open indefinitely.

3. Select, Insert, Update & Transactions

SQLAlchemy 2.0 standardizes on the select() construct for every query, whether you're using the ORM or writing Core SQL directly — the legacy session.query(Task) style still works but is considered legacy. You build a select() statement, then hand it to the session to execute:

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.models.task import Task


def find_task(session: Session, task_id: int) -> Task | None:
    return session.scalar(select(Task).where(Task.id == task_id))


def list_tasks(session: Session) -> list[Task]:
    return list(session.scalars(select(Task).order_by(Task.id)))


def create_task(session: Session, title: str, description: str | None) -> Task:
    task = Task(title=title, description=description)
    session.add(task)
    session.commit()
    session.refresh(task)
    return task


def update_task(session: Session, task: Task, *, is_done: bool) -> Task:
    task.is_done = is_done
    session.commit()
    return task

session.scalar(...) runs the statement and returns a single column value or None — ideal for "find one row by id." session.scalars(...) returns every matching row as ORM objects, which is what you iterate over for a list endpoint. Neither of these touches the database until you actually call it; building a select() object is just assembling SQL in memory.

session.add(task) doesn't write anything either — it just puts the new object under the session's tracking. The actual INSERT (or UPDATE, for changes to already-tracked objects like the is_done assignment above) happens on flush, which commit() triggers automatically. commit() is your transaction boundary: everything the session has staged since the last commit is written atomically, or none of it is. If anything raises before you reach commit(), call session.rollback() to discard the staged changes and leave the database exactly as it was — otherwise the session is left in a broken, half-flushed state that will raise confusing errors on the next query.

# Quick sanity check against a running Postgres instance
poetry run python -c "
from app.db.session import SessionLocal
from app.repositories.tasks import create_task, find_task

with SessionLocal() as session:
    t = create_task(session, 'Ship week 4', None)
    print(find_task(session, t.id))
"

4. Hands-on Exercise

5. Knowledge Check

Four quick questions. Expand each to check your answer.