Week 15: Advanced SQLAlchemy — Complex Queries, N+1 & Migrations at Scale

1. Dynamic Queries with the 2.0 select() API

A repository method like get_open_tasks(db) works when the filters are fixed. It stops working the moment a search endpoint needs to combine an arbitrary, optional subset of filters — status, assignee, and a date range, any of which might be omitted. SQLAlchemy 2.0's select() statement is a plain Python object you can build up incrementally, which makes composing it from optional conditions straightforward:

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

async def search_tasks(
    db: AsyncSession,
    status: str | None = None,
    assignee_id: int | None = None,
    created_after: datetime | None = None,
) -> list[Task]:
    stmt = select(Task)

    if status is not None:
        stmt = stmt.where(Task.status == status)
    if assignee_id is not None:
        stmt = stmt.where(Task.assignee_id == assignee_id)
    if created_after is not None:
        stmt = stmt.where(Task.created_at > created_after)

    result = await db.execute(stmt)
    return list(result.scalars().all())

Each .where() call returns a new, extended statement — nothing executes until db.execute(stmt) actually runs it — so building the query conditionally is just ordinary Python control flow, not a special query-builder DSL to learn. A caller with zero, one, or all three filters supplied gets exactly the right WHERE clause, with no combinatorial explosion of hand-written query methods to maintain.

2. Eager Loading & Solving N+1

By default, accessing a relationship on a lazily-loaded SQLAlchemy object fires a fresh query the moment you touch it — fine for one object, disastrous in a loop. Fetching 50 tasks and then accessing task.assignee on each one inside a list comprehension fires 1 query for the tasks, then 50 more, one per row, for their assignees. Eager loading tells SQLAlchemy up front which relationships to fetch alongside the main query, collapsing that into a small, fixed number of queries regardless of how many rows come back.

from sqlalchemy.orm import selectinload, joinedload

# selectinload: a SEPARATE query, but only ONE, using WHERE id IN (...)
stmt = select(Task).options(selectinload(Task.assignee))

# joinedload: a SINGLE query with a SQL JOIN
stmt = select(Task).options(joinedload(Task.assignee))

Both eliminate the N+1 pattern, but they trade off differently. joinedload fetches everything in one round trip via a LEFT JOIN, which is efficient for a many-to-one relationship (many tasks, one assignee each) but duplicates the parent row's columns once per matching child row for a one-to-many relationship, which can bloat the result set. selectinload issues a second query using WHERE assignee_id IN (...) against exactly the IDs the first query returned — still just two queries total regardless of how many tasks came back, without the row-duplication cost, which makes it the better default for one-to-many and many-to-many relationships.

3. Bulk Operations & Zero-Downtime Migrations

Calling db.add() in a loop for thousands of rows, then one commit(), still generates one INSERT statement per row under the hood. SQLAlchemy 2.0's ORM bulk operations batch that into far fewer round trips:

from sqlalchemy import insert

await db.execute(
    insert(Task),
    [{"title": row.title, "status": "open"} for row in csv_rows],
)
await db.commit()

A bulk update() follows the same idea — one statement affecting many rows, executed directly against the database rather than loading each ORM object, mutating it, and letting the unit of work issue individual UPDATE statements:

from sqlalchemy import update

await db.execute(
    update(Task)
    .where(Task.updated_at < cutoff)
    .values(status="archived")
)
await db.commit()

Like a JPA @Modifying query, this bypasses the ORM's identity map entirely — any already-loaded Task objects in the current session won't reflect the change unless explicitly refreshed or expired.

Schema changes deserve the same care. Adding a NOT NULL column with a default on a large table can lock it for the duration of the rewrite in some databases — a multi-second lock is effectively an outage for every request touching that table. The safe pattern spreads the change across separate, sequential migrations instead of one:

# Migration 1 (expand): add the column as NULLABLE -- no table lock
def upgrade():
    op.add_column("task", sa.Column("priority", sa.String(), nullable=True))

# Migration 2: application code starts writing it; backfill existing rows
def upgrade():
    op.execute("UPDATE task SET priority = 'normal' WHERE priority IS NULL")

# Migration 3 (contract): once backfilled and the app fully depends on it
def upgrade():
    op.alter_column("task", "priority", nullable=False)

No single migration in that sequence requires the old and new application code to disagree about the schema — which matters directly for a rolling deploy, where old and new instances briefly run side by side. A one-step migration that adds a NOT NULL column the app immediately depends on breaks the moment an old instance, still serving traffic mid-rollout, tries to insert a row without it.

4. Hands-on Exercise

5. Knowledge Check

Three quick questions. Expand each to check your answer.