Module 4: The N+1 problem with SQLAlchemy

Module 4 project: eliminating N+1 from the bookstore

Next →

What will you build and why?

In module 1 you met the bookstore and planted a problematic endpoint: GET /books-with-author?author_name=X. At that moment you only identified it as suspicious. You enabled echo=True, counted 51 queries per request, and said "we'll fix it later". "Later" has arrived.

In this module 4 project, you take that endpoint and tear it apart:

  • You measure the baseline with wrk (module 1) to have a solid starting number.
  • You diagnose with nplusone and echo=True (capsules 02-03) to understand exactly where the lazy loads are.
  • You apply the correct combination of selectinload + possibly joinedload + raiseload (capsules 04-07).
  • You validate that the N+1 disappeared: nplusone detects nothing, echo=True shows 2-3 fixed queries.
  • You report the improvement with numbers: queries before/after, p50/p95/p99 latency before/after.

The result is a "before/after" commit with concrete metrics you'll be able to quote verbatim in an interview: "I reduced an endpoint from 51 queries and 850ms p95 to 2 queries and 95ms p95 by applying correct selectinload in SQLAlchemy 2.0 async."

That's portfolio-worthy. It's the difference between "I worked with SQLAlchemy" and "I master SQLAlchemy performance tuning in production".

Project objective

By completing this project:

  • You'll have measured the problematic endpoint's baseline with wrk (p50/p95/p99 latency) and the query count with nplusone.
  • You'll have applied the correct eager loading techniques, justifying the choice between selectinload and joinedload for each relationship.
  • You'll have configured nplusone in pytest so CI breaks if someone re-introduces an N+1.
  • You'll have produced a before/after report with concrete numbers in a BENCHMARKS.md format.
  • You'll have committed the changes on a clean branch, ready for a PR.

How it fits with what you learned

This project integrates all the module's capsules:

CapsuleHow it's used in the project
02 — What is N+1?Initial diagnosis: count the endpoint's queries with echo=True
03 — Detecting with nplusoneConfigure nplusone in pytest to validate after the fix and prevent regressions
04 — joinedload vs selectinloadChoose the right strategy for Author → books → reviews
05 — subqueryloadNot applied directly, but you use the knowledge to rule it out
06 — Async and AsyncSessionAll the code is async; you apply canonical patterns
07 — Anti-patternsYou validate you're not over-fetching or using selectinload('*')

Think of the project as a mini case study: you take a broken endpoint, apply engineering discipline (measure, diagnose, fix, validate, report), and produce documented evidence of the impact.

Technical specifications

Stack

  • Language: Python 3.12+
  • Framework: FastAPI 0.110+
  • ORM: SQLAlchemy 2.0+ async
  • DB driver: asyncpg
  • Database: PostgreSQL 16+
  • Testing: pytest + pytest-asyncio + httpx
  • N+1 detection: nplusone
  • Load testing: wrk (installed in module 1)

Initial setup

If you have the module 1 bookstore running, skip to step 2. If not:

Step 1: bring up the module 1 bookstore.

# Assume the bookstore-baseline folder exists from module 1
cd ~/projects/bookstore-baseline

# If Postgres isn't running:
docker run -d --name bookstore-pg \
    -e POSTGRES_USER=bookstore \
    -e POSTGRES_PASSWORD=bookstore \
    -e POSTGRES_DB=bookstore \
    -p 5432:5432 \
    postgres:16

# Activate the venv:
source venv/bin/activate

# Re-run the seed if the DB is empty:
python seed.py

# Verify the problematic endpoint is still slow:
uvicorn main:app --reload &
curl "http://localhost:8000/books-with-author?author_name=tolkien" | head -c 200

Step 2: create a branch for the project.

git checkout -b module-04-eliminate-n1

Step 3: install new dependencies.

pip install nplusone pytest pytest-asyncio httpx
pip freeze > requirements.txt

Mandatory features

1. Establish the baseline

Before changing a line of code, measure the endpoint's current state.

1.1 — Query count

Enable echo=True and count:

# In main.py make sure you have:
# engine = create_async_engine(DATABASE_URL, echo=True)

uvicorn main:app --reload 2> baseline_echo.log &
curl "http://localhost:8000/books-with-author?author_name=tolkien" > /dev/null
sleep 1
# Stop uvicorn

# Count SELECTs
grep -c "INFO sqlalchemy.engine.Engine SELECT" baseline_echo.log

Note the number. For tolkien with 20 books from the default seed, you expect ~22 queries (1 author + 1 books + 20 reviews).

1.2 — Latency with wrk

Hit the endpoint with wrk to measure percentiles:

uvicorn main:app --reload &
sleep 2

wrk -t4 -c20 -d30s --latency \
    "http://localhost:8000/books-with-author?author_name=tolkien"

Typical output (will vary depending on your hardware):

Running 30s test @ http://localhost:8000/books-with-author?author_name=tolkien
  4 threads and 20 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency   245.32ms   58.14ms   650ms    78.45%
    Req/Sec    20.43      4.12     35.00    72.10%
  Latency Distribution
     50%   235.00ms
     75%   281.00ms
     90%   320.00ms
     99%   485.00ms
  2438 requests in 30.00s, 5.23MB read
Requests/sec:     81.27

Note: p50, p95, p99, RPS.

1.3 — Validate with nplusone that there's an N+1

Configure nplusone in warn mode:

# In main.py, after creating the engine:
import os

if os.getenv("APP_ENV") in ("development", "testing"):
    from nplusone.ext.sqlalchemy import NPlusOne
    NPlusOne(engine)

Hit the endpoint and observe warnings:

APP_ENV=development uvicorn main:app --reload &
curl "http://localhost:8000/books-with-author?author_name=tolkien"
# In the logs you'll see many:
# WARNING:nplusone:Potential n+1 query detected on `Book.reviews`

Confirmed: there's an N+1. You have a complete baseline.

2. Diagnosis

Look at the endpoint's current code in main.py and answer:

  • How many queries does it fire?
  • Which relationships are lazy?
  • What's the optimal strategy for each one?

Model structure (recap from module 1):

class Author(Base):
    __tablename__ = "authors"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(unique=True)
    books: Mapped[list["Book"]] = relationship(back_populates="author")


class Book(Base):
    __tablename__ = "books"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"))
    author: Mapped["Author"] = relationship(back_populates="books")
    reviews: Mapped[list["Review"]] = relationship(back_populates="book")


class Review(Base):
    __tablename__ = "reviews"
    id: Mapped[int] = mapped_column(primary_key=True)
    book_id: Mapped[int] = mapped_column(ForeignKey("books.id"))
    rating: Mapped[int]
    book: Mapped["Book"] = relationship(back_populates="reviews")

Current endpoint (with N+1):

@app.get("/books-with-author")
async def list_books_with_author(author_name: str):
    async with SessionLocal() as session:
        author_q = select(Author).where(Author.name == author_name)
        author = (await session.execute(author_q)).scalar_one_or_none()
        if not author:
            return {"error": "not found"}

        books_q = select(Book).where(Book.author_id == author.id)
        books = (await session.scalars(books_q)).all()

        result = []
        for book in books:
            # N+1: for each book, a separate query for reviews
            reviews_q = select(Review).where(Review.book_id == book.id)
            reviews = (await session.scalars(reviews_q)).all()
            result.append({
                "title": book.title,
                "review_count": len(reviews),
            })

        return {"author": author.name, "books": result}

Expected analysis:

  • 1 query for author.
  • 1 query for books.
  • N queries (one per book) for reviews. For tolkien with 20 books = 20 queries.
  • Total: 22 queries.
  • Optimal strategy:
    • Author → books: small 1:N (20 books). Could be joinedload or selectinload. Since there's another nested layer, selectinload.
    • Book → reviews: 1:N (5 reviews per book on average). selectinload to avoid a cartesian explosion.
    • Combination: selectinload(Author.books).selectinload(Book.reviews). Total: 3 queries.

3. Apply the fix

3.1 — Refactor the endpoint

from sqlalchemy.orm import selectinload, raiseload
from sqlalchemy import select, func

@app.get("/books-with-author")
async def list_books_with_author(author_name: str):
    async with SessionLocal() as session:
        stmt = (
            select(Author)
            .options(
                selectinload(Author.books).selectinload(Book.reviews),
                raiseload('*'),  # discipline: any other lazy → error
            )
            .where(Author.name == author_name)
        )
        author = await session.scalar(stmt)
        if not author:
            return {"error": "not found"}

        return {
            "author": author.name,
            "books": [
                {
                    "title": book.title,
                    "review_count": len(book.reviews),
                }
                for book in author.books
            ],
        }

Key changes:

  1. A single main query with select(Author) instead of two separate queries (author + books).
  2. selectinload(Author.books).selectinload(Book.reviews) loads both levels eagerly.
  3. raiseload('*') prevents any OTHER relationship from being loaded lazy accidentally in the future.
  4. List comprehension inside the session block — everything accessed while the session is open.

3.2 — Optional improvement with func.count

If you only need the review count (not the complete reviews), you can avoid loading the whole reviews and use SQL aggregation:

from sqlalchemy import select, func

@app.get("/books-with-author")
async def list_books_with_author(author_name: str):
    async with SessionLocal() as session:
        # Subquery for review counts by book_id
        review_count_subq = (
            select(
                Review.book_id,
                func.count(Review.id).label("review_count"),
            )
            .group_by(Review.book_id)
            .subquery()
        )

        stmt = (
            select(Author, Book, review_count_subq.c.review_count)
            .join(Book, Book.author_id == Author.id)
            .outerjoin(review_count_subq, review_count_subq.c.book_id == Book.id)
            .where(Author.name == author_name)
        )

        rows = (await session.execute(stmt)).all()
        if not rows:
            return {"error": "not found"}

        author_name_value = rows[0][0].name
        return {
            "author": author_name_value,
            "books": [
                {"title": book.title, "review_count": count or 0}
                for _, book, count in rows
            ],
        }

This version does a single query with a JOIN and aggregation. It's the most efficient — it returns only the counts, not the complete reviews. The trade-off is more complex code. For this project, the version with selectinload is the correct solution (more readable, sufficient for the expected data).

4. Post-fix validation

4.1 — Re-measure queries with echo=True

APP_ENV=development uvicorn main:app --reload 2> after_echo.log &
curl "http://localhost:8000/books-with-author?author_name=tolkien" > /dev/null
sleep 1
# Stop uvicorn
grep -c "INFO sqlalchemy.engine.Engine SELECT" after_echo.log

Expected: 3 queries. If you see more, something lazy slipped through.

4.2 — Re-measure latency with wrk

wrk -t4 -c20 -d30s --latency \
    "http://localhost:8000/books-with-author?author_name=tolkien"

Expected:

  • p50 improves 5-15x (from ~235ms to ~15-50ms).
  • p95 improves 5-15x (from ~320ms to ~30-80ms).
  • RPS increases 5-15x.

(The exact numbers depend on your hardware and the specific tolkien. What matters is the relative improvement.)

4.3 — Verify with nplusone

APP_ENV=development uvicorn main:app --reload &
curl "http://localhost:8000/books-with-author?author_name=tolkien"
# You should NOT see: WARNING:nplusone:Potential n+1...

If there are no warnings, the N+1 is eliminated.

5. Automated tests

Configure pytest with nplusone in raise mode so CI breaks if someone re-introduces an N+1.

5.1 — tests/conftest.py

import os
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from nplusone.ext.sqlalchemy import NPlusOne

os.environ["NPLUSONE_RAISE"] = "True"

from main import app

TEST_DB_URL = "postgresql+asyncpg://bookstore:bookstore@localhost:5432/bookstore"
test_engine = create_async_engine(TEST_DB_URL, echo=False)
NPlusOne(test_engine)


@pytest_asyncio.fixture
async def client():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac

5.2 — tests/test_books_with_author.py

import pytest


@pytest.mark.asyncio
async def test_books_with_author_no_n_plus_one(client):
    """
    If someone re-introduces a lazy load in /books-with-author,
    nplusone raises NPlusOneError and this test fails.
    """
    response = await client.get("/books-with-author?author_name=tolkien")
    assert response.status_code == 200
    data = response.json()
    assert data["author"] == "tolkien"
    assert len(data["books"]) > 0
    # Each book must have title and review_count
    for book in data["books"]:
        assert "title" in book
        assert "review_count" in book
        assert isinstance(book["review_count"], int)


@pytest.mark.asyncio
async def test_books_with_author_not_found(client):
    response = await client.get("/books-with-author?author_name=nonexistent")
    # Behavior: depends on how you handle the error in your endpoint
    assert response.status_code in (200, 404)

5.3 — Run the tests

pytest tests/ -v

Expected: both tests pass. If test_books_with_author_no_n_plus_one fails with NPlusOneError, there's still a lazy load. Diagnose it and fix it.

6. Benchmark report

Create BENCHMARKS.md at the project root with the format:

# Benchmarks: /books-with-author

## Setup

- **Hardware:** [your CPU, RAM, OS]
- **PostgreSQL:** 16.x
- **Python:** 3.12.x
- **SQLAlchemy:** 2.0.x
- **wrk:** -t4 -c20 -d30s
- **Target:** /books-with-author?author_name=tolkien
- **Seed data:** 5,000 authors, 100,000 books, 500,000 reviews
- **tolkien:** 20 books with ~5 reviews each

## Results

| Metric | Before (with N+1) | After (with selectinload) | Improvement |
|---------|------------------|---------------------------|--------|
| SQL queries per request | 22 | 3 | **7.3x fewer** |
| p50 latency | 235ms | 18ms | **13x faster** |
| p95 latency | 320ms | 42ms | **7.6x faster** |
| p99 latency | 485ms | 78ms | **6.2x faster** |
| Sustained RPS | 81 | 612 | **7.5x more throughput** |

## Diagnosis

The endpoint fired 1 query for author + 1 query for books + N queries for reviews
(one per book). For tolkien with 20 books, total: 22 queries in series.

Each individual query was fast (~0.5ms in pure SQL), but the network round-trip cost
added up: 22 × ~10ms = 220ms just in transport per request.

## Applied solution

Replaced `for book in books: query_reviews(book.id)` with:

```python
stmt = (
    select(Author)
    .options(selectinload(Author.books).selectinload(Book.reviews))
    .where(Author.name == author_name)
)

This reduces to 3 fixed queries regardless of the number of books or reviews.

Regression checks

  • pytest test with NPLUSONE_RAISE=True validates that no N+1 appears.
  • If someone edits the endpoint and leaves a lazy load, CI breaks.
  • raiseload('*') in the query prevents accidentally loading another relationship.

Next steps (module 8)

  • Apply the same techniques to other bookstore endpoints.
  • Profiling with pg_stat_statements (module 5) to find non-obvious N+1s.
  • Connection pooling with PgBouncer (module 6) to sustain higher RPS.

**The number of columns and metrics are a guide — adapt them to your real measurements.**

### 7. Commit and PR

```bash
git add main.py tests/ BENCHMARKS.md requirements.txt
git commit -m "Eliminate N+1 in /books-with-author with nested selectinload"
git push origin module-04-eliminate-n1

If you have a GitHub repo, open a PR with a description that includes the content of BENCHMARKS.md. That's portfolio material: anyone who reads it sees what you did, why, and what improved.

Validations and error handling

What must be validated

  • The endpoint still works: status 200 with tolkien.
  • The response still has the same format: {"author": ..., "books": [{"title": ..., "review_count": ...}]}.
  • For a nonexistent author, behavior consistent with before (404 or {"error": "not found"}).
  • The number of SQL queries in echo=True is constant (3) no matter how many books the author has.
  • The pytest test with NPLUSONE_RAISE=True passes.
  • The p95 latency improves at least 5x.

Errors that must be handled

  • Author not found: return 404 or a clear error structure.
  • Author with no books: return {"books": []}, not an error.
  • DB unavailable: connection errors propagate as 500 (DB error handling is module 6).

Minimal implementation example

Complete post-fix main.py:

import os
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import selectinload, raiseload

from models import Author, Book, Review

DATABASE_URL = "postgresql+asyncpg://bookstore:bookstore@localhost:5432/bookstore"

engine = create_async_engine(DATABASE_URL, echo=False)
SessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

# Enable nplusone only in dev/testing
ENV = os.getenv("APP_ENV", "production")
if ENV in ("development", "testing"):
    from nplusone.ext.sqlalchemy import NPlusOne
    NPlusOne(engine)


@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    await engine.dispose()


app = FastAPI(lifespan=lifespan)


@app.get("/health")
async def health():
    return {"status": "ok"}


@app.get("/books-with-author")
async def list_books_with_author(author_name: str):
    async with SessionLocal() as session:
        stmt = (
            select(Author)
            .options(
                selectinload(Author.books).selectinload(Book.reviews),
                raiseload('*'),
            )
            .where(Author.name == author_name)
        )
        author = await session.scalar(stmt)
        if not author:
            raise HTTPException(404, f"Author '{author_name}' not found")

        return {
            "author": author.name,
            "books": [
                {
                    "title": book.title,
                    "review_count": len(book.reviews),
                }
                for book in author.books
            ],
        }

To run:

APP_ENV=development uvicorn main:app --reload

Evaluation rubric (self-check)

Diagnosis (20 points)

  • (5 pts) You enabled echo=True and counted baseline queries.
  • (5 pts) You configured nplusone and confirmed N+1 warnings.
  • (5 pts) You measured baseline latency with wrk (p50, p95, p99, RPS).
  • (5 pts) You identified exactly which lazy loading was in the code.

Implementation (35 points)

  • (10 pts) You refactored the endpoint with correct nested selectinload.
  • (5 pts) You added raiseload('*') to prevent regressions.
  • (5 pts) The refactor preserves exactly the same response format.
  • (5 pts) You validated post-fix that echo=True shows 3 fixed queries.
  • (5 pts) You validated post-fix that nplusone emits no warnings.
  • (5 pts) You configured expire_on_commit=False correctly.

Automated tests (15 points)

  • (5 pts) tests/conftest.py configures nplusone in raise mode.
  • (5 pts) A test exercises the endpoint and validates that it passes.
  • (5 pts) A test covers the author-not-found case.

Benchmarking (15 points)

  • (5 pts) You measured post-fix latency with wrk (same setup as the baseline).
  • (5 pts) You calculated the relative improvement (X times faster) per percentile.
  • (5 pts) You documented the improvement in sustained RPS.

Documentation (15 points)

  • (10 pts) BENCHMARKS.md with a before/after table, diagnosis, solution.
  • (5 pts) Descriptive commit message (not just "fix N+1").

Extra credit (optional, up to +10 pts)

  • (+5 pts) You implemented the alternative version with func.count and an explicit JOIN; you compared latency with the selectinload version.
  • (+5 pts) You applied the pattern to another bookstore endpoint with N+1 (create one if it doesn't exist).

Total: 100 points. Passing: ≥75 / 100. Excellent: ≥90 / 100 — ready to show in an interview.

Common mistakes in this project

Mistake 1: measuring latency with the server in debug mode

Symptom: "My baseline p95 is 800ms, after the fix it's still 700ms."

Why it happens: you ran wrk against uvicorn --reload, which has file watching and reload overhead. The measurements are noisy.

How to fix it: run without --reload to measure:

uvicorn main:app --workers 1

And measure with wrk only after the server is stable (1-2 seconds).

Mistake 2: forgetting expire_on_commit=False

Symptom: after the refactor, you see random MissingGreenlet on some requests.

Why it happens: expire_on_commit=True (default) invalidates objects after a commit. In async, that leads to accidental lazy loads that fail.

How to fix it:

SessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,  # ← essential in async
)

Mistake 3: tests without pytest-asyncio configured

Symptom: pytest shows "skipped" on async tests, or fails with "fixture not found".

Why it happens: pytest-asyncio needs explicit configuration.

How to fix it: add pytest.ini:

[pytest]
asyncio_mode = auto

Or pyproject.toml:

[tool.pytest.ini_options]
asyncio_mode = "auto"

Mistake 4: NPLUSONE_RAISE isn't applied

Symptom: "My test doesn't fail even though there's an N+1."

Why it happens: the env var is set after importing NPlusOne, or it's set in the shell but pytest doesn't inherit it.

How to fix it: configure it BEFORE any nplusone import:

# tests/conftest.py — at the start of the file
import os
os.environ["NPLUSONE_RAISE"] = "True"

# Now you can import
from nplusone.ext.sqlalchemy import NPlusOne

Or use pytest-env with pytest.ini:

[pytest]
env =
    NPLUSONE_RAISE=True

Mistake 5: forgetting raiseload('*') and introducing a regression

Symptom: weeks after the fix, someone edits the endpoint, adds book.author.country to the response. N+1 comes back.

Why it happens: without raiseload('*'), new lazy accesses don't break or alert.

How to prevent it: raiseload('*') in the initial query. Any future lazy access raises InvalidRequestError with a clear message. CI breaks. Bug caught before prod.

Mistake 6: comparing measurements from different hardware

Symptom: "My colleague measured a p95 of 100ms on their laptop, I measure 300ms."

Why it happens: different hardware, different seed data distribution, other processes consuming CPU.

How to handle it: what matters is the relative improvement (from X to Y, factor Z), not the absolute numbers. Each person measures in their own environment with their own baseline.

What to do if you get stuck

  • Setup doesn't work: review module 1 capsule 08 (baseline project). Make sure the bookstore runs and has seeded data.
  • nplusone doesn't detect an obvious N+1: verify that you passed the correct engine to NPlusOne(engine). Enable echo=True to see the SQL — if you see queries in a loop, there's an N+1 even if nplusone doesn't warn.
  • MissingGreenlet post-fix: you're accessing an unloaded relationship. Capture echo=True, identify which relationship, and add it to the selectinload. Or raiseload('*') will tell you exactly which one.
  • The test passes but wrk doesn't improve: check that you're measuring the correct endpoint and that the server restarted with your changes. Basic but common mistakes.
  • Post-fix latency the same as baseline: verify that selectinload was really applied (3 queries in echo=True). If you see 22 queries, the refactor didn't take due to some import error.

What's next

What you built here is the foundation of the module 8 consolidating final project. There you'll take the bookstore with ALL the problems (not just N+1, but also large OFFSET, COUNT(*), missing GIN, untuned pool) and solve each one by applying techniques from modules 1-8. The N+1 will already be resolved because you fixed it here.

Before advancing to module 5, make sure your project meets:

  • The /books-with-author endpoint fires exactly 3 queries (verifiable with echo=True).
  • The pytest test with NPLUSONE_RAISE=True passes.
  • BENCHMARKS.md documents the improvement with before/after numbers.
  • The branch is committed and pushed (even if you don't open a PR yet).

Next module (5): Query Profiling in Production. Your current workflow depends on echo=True in dev. In production you can't enable that. You learn pg_stat_statements to identify your app's most expensive queries without touching code, and auto_explain to capture slow query plans automatically. You'll take this module's detection skills (where echo=True shows you everything) to production (where you only have aggregated metrics and have to recognize patterns).

Resources for the project

  1. SQLAlchemy 2.0 — Relationship Loading Techniques — reference for all the eager loading options.
  2. jmcarp/nplusone (GitHub) — the detection library.
  3. wrk — A modern HTTP benchmarking tool (GitHub) — the load testing tool.
  4. FastAPI — Testing — pytest with FastAPI fundamentals.
  5. pytest-asyncio Documentation — configuration for async tests.
  6. Mike Bayer — "Asynchronous I/O in SQLAlchemy" — the official async reference for the project's patterns.
  7. Asif Muhammad — "Solving the N+1 problem in FastAPI with SQLAlchemy eager loading" — a case applied to the path's stack with complete code.

Module 4 — Database Performance & Query Tuning Guide

← PreviousEager loading anti-patterns: when the cure is worseNext →Module 5: Query Profiling in Production