Module 4: The N+1 problem with SQLAlchemy

Detecting N+1 automatically: logs and the `nplusone` library

Capsule overview

In the previous capsule you learned to count queries with echo=True and to predict N+1 by reading code. That's enough to diagnose a specific endpoint when you already suspect something is wrong. It's not enough for a production app with 80 endpoints where any merge can introduce a new N+1 without anyone noticing.

What you need is automatic detection integrated into your pipeline:

  • In tests: when a PR introduces an N+1, the tests fail. CI breaks, the reviewer sees the error, it doesn't get merged. The N+1 never reaches production.
  • In dev: when you run the local app, a visible warning appears in the console every time an endpoint fires an N+1. You see it immediately, without having to enable echo=True and read logs manually.

The library that does both is called nplusone. This capsule teaches you to:

  • Install and configure nplusone with SQLAlchemy 2.0.
  • Integrate it in pytest so it raises on any new N+1.
  • Integrate it as FastAPI middleware so it warns in dev.
  • Differentiate legitimate false positives from real N+1s.
  • Build a simpler detector from scratch with SQLAlchemy event listeners (for cases where nplusone doesn't fit).

By the end, your CI rejects PRs with new N+1s, and your dev workflow shouts when one appears.


Mental model: the detector as guardian of the workflow

Think of nplusone as a smoke detector in your kitchen:

  • It doesn't cook for you.
  • It doesn't tell you how to put out the fire.
  • But it warns you instantly when something catches fire, instead of you finding out when there are already 50 people inside the restaurant (production).

The equivalent in code:

  • nplusone doesn't solve the N+1 (that's joinedload / selectinload, capsules 04-05).
  • It doesn't teach you to choose the strategy.
  • But it warns you instantly when a lazy access happens, while you run tests or browse dev — before the end user hits the endpoint in production.

Like a smoke detector, there are two modes of operation:

  1. raise mode (in tests): any lazy access = immediate exception. Test fails. It's the "angry guard" mode that lets nothing through.
  2. warn mode (in dev): any lazy access = printed warning. The app keeps working, but you see the notice. It's the "attentive assistant" mode that reminds you to check.

In both cases, nplusone integrates as a middleware (in FastAPI) or as a fixture (in pytest) that surrounds each request or each test, monitors accesses to relationships, and fires the configured action when it detects a lazy load.


Installation and base configuration

pip install nplusone

nplusone supports several ORMs (SQLAlchemy, Django, Peewee). We use the SQLAlchemy module.

Initial configuration

nplusone "hooks" into the SQLAlchemy engine by listening to events. Activation is a single line:

from nplusone.ext.sqlalchemy import NPlusOne

# In the app setup
NPlusOne(engine)

After that line, nplusone starts observing all accesses to relationships of the passed engine. If it detects that a lazy attribute was loaded after the parent query already finished, it fires its configured action.

Relevant configuration variables

nplusone is configured with env vars or programmatically:

VariableDefaultDescription
NPLUSONE_LOGGERnplusone (Python logger)Which logger it emits warnings to
NPLUSONE_LOG_LEVELWARNINGLogging level
NPLUSONE_RAISEFalseIf True, raises NPlusOneError instead of warning
NPLUSONE_WHITELIST[]List of relationships to ignore (legitimate false positives)

Canonical pattern:

  • In dev: NPLUSONE_RAISE=False (warn).
  • In tests: NPLUSONE_RAISE=True (raise).
  • In production: nplusone is not loaded. It has overhead you don't want in the hot path.

Integration in pytest: raise mode

Tests are where nplusone gives you the most value. Configured in raise mode, any N+1 that a test exercises blows up as NPlusOneError. CI breaks, you see the stack trace, you know exactly which relationship was loaded lazy.

Complete setup

Project structure:

my-app/
├── app/
│   ├── main.py
│   ├── models.py
│   └── db.py
├── tests/
│   ├── conftest.py
│   ├── test_endpoints.py
│   └── ...
├── pytest.ini
└── requirements.txt

tests/conftest.py:

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

from app.main import app
from app.db import get_db
from app.models import Base


# Enable nplusone in raise mode for all tests
os.environ["NPLUSONE_RAISE"] = "True"

# Test engine against a real DB (use testcontainers or a dedicated test DB)
TEST_DB_URL = "postgresql+asyncpg://test:test@localhost:5432/bookstore_test"
test_engine = create_async_engine(TEST_DB_URL, echo=False)

# Enable nplusone after creating the engine
NPlusOne(test_engine)

TestSessionLocal = async_sessionmaker(test_engine, expire_on_commit=False)


@pytest_asyncio.fixture(scope="function")
async def db_session():
    """DB session with automatic rollback at the end of the test."""
    async with TestSessionLocal() as session:
        yield session
        await session.rollback()


@pytest_asyncio.fixture(scope="function")
async def client(db_session):
    """Async client to make requests to the endpoint."""
    async def override_get_db():
        yield db_session

    app.dependency_overrides[get_db] = override_get_db
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac

    app.dependency_overrides.clear()


@pytest_asyncio.fixture(scope="session", autouse=True)
async def setup_database():
    """Creates the schema before the suite, cleans up at the end."""
    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

pytest.ini:

[pytest]
asyncio_mode = auto
testpaths = tests

Test that catches N+1

# tests/test_endpoints.py
import pytest
from sqlalchemy import select
from app.models import Author, Book


@pytest.mark.asyncio
async def test_books_with_author_no_n_plus_one(client, db_session):
    """
    This test fails if the /books-with-author endpoint has an N+1.
    """
    # Setup: create minimal data
    author = Author(name="test_author")
    db_session.add(author)
    await db_session.flush()

    for i in range(5):
        db_session.add(Book(title=f"book_{i}", author_id=author.id))
    await db_session.commit()

    # Run the endpoint. If there's an N+1 loading relationships,
    # nplusone raises NPlusOneError and the test fails.
    response = await client.get("/books-with-author?author_name=test_author")

    assert response.status_code == 200
    data = response.json()
    assert data["author"] == "test_author"
    assert len(data["books"]) == 5

When the test fails

If the endpoint has an N+1, the test fails with a message like this:

FAILED tests/test_endpoints.py::test_books_with_author_no_n_plus_one
nplusone.core.exceptions.NPlusOneError:
Potential n+1 query detected on `Book.reviews`

nplusone tells you which relationship was loaded lazy. You go to the endpoint code, add selectinload(Book.reviews), run the test again. It passes. PR approved.

Incremental adoption strategy

If you have an existing app with N+1s already in production, enabling nplusone raise breaks all the tests at once. Gradual strategy:

  1. Start with NPLUSONE_RAISE=False in tests, warnings log only.
  2. Capture the warnings in a file (pytest --capture=no | tee nplusone_warnings.txt).
  3. List all the relationships that appear.
  4. Fix the top-K most common ones with eager loading.
  5. Mark the rest as whitelist (the capsule explains below) if they're acceptable.
  6. Once whitelist + fixes = green tests, switch to NPLUSONE_RAISE=True.
  7. From then on, any new N+1 breaks CI.

Integration in FastAPI: middleware in warn mode

For dev, you want nplusone to warn you when making requests without breaking the app. You integrate it as ASGI middleware.

Setup

nplusone comes with a specific middleware for ASGI/WSGI frameworks. For FastAPI you use the general wrapper:

# app/main.py
import os
from fastapi import FastAPI
from sqlalchemy.ext.asyncio import create_async_engine
from nplusone.ext.sqlalchemy import NPlusOne


# Only enable nplusone in dev/test, NEVER in production
ENV = os.getenv("APP_ENV", "production")

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

if ENV in ("development", "testing"):
    NPlusOne(engine)
    # In dev we want warn (default), not raise
    os.environ["NPLUSONE_RAISE"] = "False"

app = FastAPI()


# Rest of the app...

When you run APP_ENV=development uvicorn app.main:app, each lazy access emits a warning to stderr:

WARNING:nplusone:Potential n+1 query detected on `Author.books`

Configurable logging

If you want a custom format or output to a file:

import logging

logger = logging.getLogger("nplusone")
logger.setLevel(logging.WARNING)

handler = logging.FileHandler("nplusone.log")
handler.setFormatter(logging.Formatter(
    "%(asctime)s [%(levelname)s] %(message)s"
))
logger.addHandler(handler)

Now the warnings go to nplusone.log. Useful if your stdout is saturated with other logs.

Why NOT in production

nplusone installs event listeners that run Python code on each access to a relationship. It's small but measurable overhead. On hot endpoints (high RPS) it can add significant latency.

More importantly: in production, if an N+1 occurs (because of something that slipped through despite your CI), you don't want the app to raise an exception at the user. You want the N+1 to manifest as latency (which you'll detect with pg_stat_statements, module 5), not as a 500.

Simple rule: nplusone enabled only when APP_ENV in ("development", "testing").


Whitelist: handling false positives

nplusone sometimes detects lazy accesses that are intentional and acceptable:

  • Loading a single field on demand (not in a loop). Example: in GET /authors/{id}, bringing 1 author and then accessing author.books ONCE.
  • Small relationships always loaded in mappings with default eager. Example: lazy="joined" in the model, but nplusone still reports because it detects it as a post-query access.
  • Polymorphic relationships with special loading that nplusone doesn't understand.

For those cases, configure a whitelist:

import os

os.environ["NPLUSONE_WHITELIST"] = '[{"label": "n_plus_one", "model": "Author", "field": "books"}]'

Or programmatically:

from nplusone.core.notifiers import on_event

@on_event
def custom_notifier(event):
    # Ignore accesses to Author.books if the endpoint is /authors/{id}
    if event.objects[0].__class__.__name__ == "Author" and event.field == "books":
        # Only log, don't raise
        return
    # ...

Careful with the whitelist: each entry is technical debt. Annotate it with a reason in a comment:

# WHITELIST: Author.books in /authors/{id}/profile.
# It's 1 author + 1 lazy load (no loop). Acceptable.

If the whitelist grows out of control, you lose the tool's trust. Review it every sprint.


Custom detector with event listeners (no library)

Sometimes you don't want additional dependencies (restrictive projects, a lib out of date for your SQLAlchemy version). Building a simple detector with SQLAlchemy event listeners is viable.

Basic pattern: count queries per request

# app/middleware/query_counter.py
from contextvars import ContextVar
from sqlalchemy import event
from sqlalchemy.ext.asyncio import AsyncEngine

# ContextVar to keep the per-request count in async
query_count_var: ContextVar[int] = ContextVar("query_count", default=0)


def setup_query_counter(engine: AsyncEngine):
    """Installs a listener that increments the counter on each query."""
    @event.listens_for(engine.sync_engine, "before_cursor_execute")
    def count_query(conn, cursor, statement, parameters, context, executemany):
        current = query_count_var.get()
        query_count_var.set(current + 1)


class QueryCountMiddleware:
    """ASGI middleware that logs queries per request."""
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        # Reset the counter at the start of the request
        query_count_var.set(0)

        await self.app(scope, receive, send)

        # Log the total at the end
        total = query_count_var.get()
        path = scope.get("path", "")
        if total > 10:  # arbitrary threshold to alert
            print(f"[QUERY-WARN] {path} fired {total} queries")
        else:
            print(f"[QUERY-OK] {path} fired {total} queries")

Integration in FastAPI

from fastapi import FastAPI
from app.db import engine
from app.middleware.query_counter import setup_query_counter, QueryCountMiddleware

app = FastAPI()
setup_query_counter(engine)
app.add_middleware(QueryCountMiddleware)

Each request now shows:

[QUERY-OK] /authors/42 fired 1 queries
[QUERY-WARN] /books-with-author?author_name=tolkien fired 52 queries

Advantages vs nplusone:

  • Zero additional dependencies.
  • Total granularity: you decide the threshold, the format, the destination.

Disadvantages:

  • It doesn't differentiate "planned queries" from "lazy loads" — it only counts.
  • You need to keep the count manually with ContextVar (tricky in async).
  • It requires maintenance if the SQLAlchemy API changes.

For most cases, nplusone is the right choice. The custom detector is useful for edge cases or one-off audits.


Integrating into CI: GitHub Actions workflow

Canonical CI pattern with N+1 detection:

# .github/workflows/test.yml
name: Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16
        env:
          POSTGRES_USER: test
          POSTGRES_PASSWORD: test
          POSTGRES_DB: bookstore_test
        ports:
          - 5432:5432
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5

    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install pytest pytest-asyncio nplusone httpx

      - name: Run tests with N+1 detection
        env:
          NPLUSONE_RAISE: "True"
          DATABASE_URL: postgresql+asyncpg://test:test@localhost:5432/bookstore_test
        run: pytest -v

After merging this config: any PR that introduces a new N+1 in an endpoint with an associated test fails CI with a clear message. Your team doesn't merge that PR. The N+1 never goes out to the real world.


Why this matters in real work

1. CI with N+1 detection eliminates an entire class of bugs.

Without this, N+1s enter production and are discovered with dashboards that show a high p95. Diagnosis: hours reading the APM, comparing spans, suspecting endpoints. With this, the N+1 is discovered in CI with an exact stack trace. Diagnosis: 30 seconds.

2. It's scalable discipline.

If your team grows from 3 to 30 devs, you can't eyeball every PR to detect N+1. Automation is the only way to keep the standard.

3. The warning in dev changes the feedback loop.

Without a warning, the dev writes code that looks fine and discovers the N+1 when some test fails much later (or never). With a warning, they see it instantly in their terminal. They learn the pattern immediately.

4. It sets you apart in a technical interview.

"How do you make sure you don't introduce N+1s?" — junior answer: "I look at the logs". Senior answer: "we have nplusone configured in pytest with raise, and as middleware in dev with warn. Any new N+1 breaks CI before review."


Traps and common mistakes

Mistake 1 (configuration): enabling nplusone in production

Symptom: "My p95 went up 20ms after a deploy."

Why it happens: you forgot the if ENV != "production" guard. nplusone is running event listeners on each access to a relationship in the hot path.

How to distinguish: check prod logs looking for "nplusone" or the loaded module. Confirm the env var.

How to fix it: strict pattern:

if os.getenv("APP_ENV") in ("development", "testing"):
    NPlusOne(engine)

And in CI, make sure APP_ENV is not "production" for the test steps.

Mistake 2 (configuration): NPLUSONE_RAISE in dev breaks the app

Symptom: "Every time I hit the endpoint in dev I get a 500 error."

Why it happens: you enabled RAISE=True by mistake. In dev you want warn, in tests raise.

How to fix it: verify active env vars:

env | grep NPLUSONE

Switch to RAISE=False in the dev .env. Only True in the tests .env or in CI.

Mistake 3 (conceptual): assuming nplusone fixes N+1

Symptom: "I enabled nplusone but the queries are still many."

Why it happens: nplusone only detects. The solution is joinedload/selectinload (capsules 04-05).

How to distinguish: if your CI breaks with "n+1 detected" and your fix is "add the selectinload option in the query", you're using nplusone correctly.

Mistake 4 (practical): tests pass locally but fail in CI

Symptom: "Locally with pytest everything passes, in GitHub Actions it fails with NPlusOneError."

Why it happens: you forgot to configure NPLUSONE_RAISE=True locally too, so locally you only saw warnings (which get lost in the noise) and in CI it does break.

How to fix it: align the local test env vars with CI. Pattern: use pytest.ini or pyproject.toml to set vars when pytest starts:

# pytest.ini
[pytest]
env =
    NPLUSONE_RAISE=True

(Requires pytest-env.)

Mistake 5 (abusive whitelist)

Symptom: "My whitelist has 30 entries, nobody understands it anymore."

Why it happens: every time a test fails, the dev adds to the whitelist instead of fixing the N+1.

How to fix it: treat each whitelist entry as a TODO. Mandatory comment in code:

# WHITELIST: Author.books — pending refactor in JIRA-1234

Review the whitelist in each team retro. Old entries are technical debt that cost dev velocity. Remove them.

Mistake 6 (interpretation): false positive on lazy="joined" or lazy="selectin"

Symptom: "I set lazy='joined' in the model and nplusone still reports N+1."

Why it happens: nplusone is based on heuristics. Sometimes it detects post-query accesses as lazy even though the relationship is preloaded with default eager loading.

How to distinguish: enable echo=True and verify whether there really are additional queries. If you only see the main query with a JOIN, it's a false positive.

How to fix it: add the relationship to the whitelist explicitly, with a comment explaining that it's a false positive due to lazy="joined".


Exercises

Exercise 1: configure nplusone in pytest

Take the module 1 bookstore. Add nplusone in raise mode to conftest.py. Write a test that exercises /books-with-author?author_name=tolkien and watch it fail with NPlusOneError.

See solution

1. Install dependencies:

pip install pytest pytest-asyncio nplusone httpx

2. Create 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 app.main import app
from app.db import get_db

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

TestSessionLocal = async_sessionmaker(test_engine, expire_on_commit=False)


@pytest_asyncio.fixture
async def db_session():
    async with TestSessionLocal() as session:
        yield session


@pytest_asyncio.fixture
async def client(db_session):
    async def override_get_db():
        yield db_session
    app.dependency_overrides[get_db] = override_get_db
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac
    app.dependency_overrides.clear()

3. Create tests/test_n_plus_one.py:

import pytest


@pytest.mark.asyncio
async def test_books_with_author_no_n_plus_one(client):
    response = await client.get("/books-with-author?author_name=tolkien")
    assert response.status_code == 200

4. Run:

pytest tests/test_n_plus_one.py -v

Expected output:

FAILED tests/test_n_plus_one.py::test_books_with_author_no_n_plus_one
nplusone.core.exceptions.NPlusOneError:
Potential n+1 query detected on `Book.reviews`

If your test fails with that message, everything is configured correctly. The next step (capsule 04) is fixing the endpoint with eager loading so the test passes.

Exercise 2: integrate nplusone as FastAPI middleware

Configure nplusone so that in dev (variable APP_ENV=development) it prints warnings to stderr every time it detects a lazy load. Verify with curl that the warning appears.

See solution

Modify app/main.py:

import os
from fastapi import FastAPI
from sqlalchemy.ext.asyncio import create_async_engine
from nplusone.ext.sqlalchemy import NPlusOne

ENV = os.getenv("APP_ENV", "production")

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

if ENV in ("development", "testing"):
    NPlusOne(engine)
    os.environ.setdefault("NPLUSONE_RAISE", "False")
    print(f"[INFO] nplusone active in mode: {os.environ['NPLUSONE_RAISE']}")

app = FastAPI()
# ... rest of the app

Start the server:

APP_ENV=development uvicorn app.main:app --reload

In another terminal:

curl "http://localhost:8000/books-with-author?author_name=tolkien" > /dev/null

In the server terminal you'll see:

[INFO] nplusone active in mode: False
WARNING:nplusone:Potential n+1 query detected on `Book.reviews`
WARNING:nplusone:Potential n+1 query detected on `Book.reviews`
... (one for each lazy access)

Also verify that with APP_ENV=production (default) it does NOT activate:

unset APP_ENV
uvicorn app.main:app --reload
# It doesn't print the [INFO] for nplusone active

Exercise 3: custom detector with an event listener

Without using nplusone, write a FastAPI middleware that counts queries per request and emits a warning if it exceeds 10. Test it against /books-with-author.

See solution

app/middleware/query_counter.py:

import logging
from contextvars import ContextVar
from sqlalchemy import event
from sqlalchemy.ext.asyncio import AsyncEngine

logger = logging.getLogger("query_counter")
logger.setLevel(logging.INFO)

query_count_var: ContextVar[int] = ContextVar("query_count", default=0)


def setup_query_counter(engine: AsyncEngine):
    @event.listens_for(engine.sync_engine, "before_cursor_execute")
    def count_query(conn, cursor, statement, parameters, context, executemany):
        try:
            current = query_count_var.get()
            query_count_var.set(current + 1)
        except LookupError:
            pass  # Outside a request context


class QueryCountMiddleware:
    def __init__(self, app, threshold: int = 10):
        self.app = app
        self.threshold = threshold

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        token = query_count_var.set(0)
        try:
            await self.app(scope, receive, send)
        finally:
            total = query_count_var.get()
            path = scope.get("path", "")
            if total > self.threshold:
                logger.warning(f"{path} -> {total} queries (threshold: {self.threshold})")
            else:
                logger.info(f"{path} -> {total} queries")
            query_count_var.reset(token)

Integrate into app/main.py:

from app.middleware.query_counter import setup_query_counter, QueryCountMiddleware

setup_query_counter(engine)
app.add_middleware(QueryCountMiddleware, threshold=10)

Enable logging:

import logging
logging.basicConfig(level=logging.INFO)

Test:

curl "http://localhost:8000/books-with-author?author_name=tolkien"

Expected output:

WARNING:query_counter:/books-with-author -> 22 queries (threshold: 10)

Versus a healthy endpoint:

curl "http://localhost:8000/health"
# INFO:query_counter:/health -> 0 queries

Difference from nplusone:

  • Your detector only counts. It doesn't distinguish planned queries from lazy loads.
  • Simpler, less precise.
  • Useful as a general tripwire; less useful for fine diagnosis.

For a quick audit without adding dependencies, it's worth it. For strict CI, nplusone is still superior.

Exercise 4: distinguish a false positive from a real N+1

Your test fails with:

NPlusOneError: Potential n+1 query detected on `User.profile`

Investigate the endpoint:

@app.get("/users/{user_id}")
async def get_user(user_id: int, session: AsyncSession = Depends(get_db)):
    user = await session.get(User, user_id)
    return {
        "name": user.name,
        "profile": {
            "bio": user.profile.bio,  # ← lazy access
        }
    }

Is it a real N+1 or a false positive? Justify it.

See solution

It's a false positive (with a nuance).

Analysis:

  • The endpoint returns 1 user (not a list).
  • There's 1 lazy access to user.profile.
  • Total: 1 parent query + 1 lazy load = 2 queries.

This is NOT structural N+1. The number of queries doesn't scale with data — it's a fixed 2. The name "N+1" requires a loop with a variable N.

However, it's inefficient:

  • 2 queries in series when they could be 1 with joinedload(User.profile).
  • Each query adds a round-trip to the server.

Correct decision:

  1. To silence nplusone specifically here, add a whitelist with a comment:
# WHITELIST: User.profile in /users/{user_id}.
# 1 user + 1 lazy load. Not structural N+1.
# TODO: optimize with joinedload (not urgent).
  1. Better: fix it with joinedload. It's not an attack on the N+1 but a general improvement:
user = await session.scalar(
    select(User)
    .options(joinedload(User.profile))
    .where(User.id == user_id)
)

Now 1 query with a LEFT JOIN. Zero N+1 alerts. Zero whitelist. Faster.

Lesson: nplusone sometimes shouts in cases that aren't strict N+1. Your job is to decide: whitelist (if it's acceptable as is) or eager (if it's worth optimizing). The only thing you shouldn't do is ignore it without thinking.

Exercise 5: configuration for CI with GitHub Actions

Write a .github/workflows/test.yml workflow that:

  1. Starts PostgreSQL 16.
  2. Installs the project dependencies.
  3. Runs pytest with NPLUSONE_RAISE=True.
See solution

.github/workflows/test.yml:

name: Tests with N+1 detection

on:
  push:
    branches: [main]
  pull_request:

jobs:
  test:
    runs-on: ubuntu-latest

    services:
      postgres:
        image: postgres:16
        env:
          POSTGRES_USER: test
          POSTGRES_PASSWORD: test
          POSTGRES_DB: bookstore_test
        ports:
          - 5432:5432
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5

    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"
          cache: "pip"

      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt
          pip install pytest pytest-asyncio nplusone httpx

      - name: Run migrations / seed
        env:
          DATABASE_URL: postgresql+asyncpg://test:test@localhost:5432/bookstore_test
        run: python -m app.seed  # your minimal seed script

      - name: Run tests with N+1 detection enabled
        env:
          NPLUSONE_RAISE: "True"
          APP_ENV: testing
          DATABASE_URL: postgresql+asyncpg://test:test@localhost:5432/bookstore_test
        run: pytest -v --tb=short

Verification:

  1. Push a branch that introduces a new N+1 (for example, an endpoint with a loop over relationships).
  2. Open a PR.
  3. The workflow runs. It fails with NPlusOneError.
  4. The PR isn't merged until it's fixed.

That's the desired result: the N+1 never reaches main.

Exercise 6: apply it to your own project

Take a FastAPI app of your own (or the bookstore if you don't have another). Configure nplusone in pytest. Run the suite. Document:

  1. How many N+1s nplusone detects.
  2. How many are false positives vs real N+1s.
  3. Which you'd fix first (prioritizing by the endpoint with the most traffic).
See solution

There's no single solution. Analysis structure:

## N+1 audit: [project name]

**Setup:**
- Configuration used: NPLUSONE_RAISE=True in pytest
- Total tests: N
- Tests that failed due to N+1: M

**N+1s detected:**

| Endpoint | Relationship | Type | Action |
|----------|--------------|------|--------|
| /authors | Author.books | Real N+1 (loop) | Fix with selectinload (P1) |
| /books/{id} | Book.reviews | False positive (1 entity) | Whitelist with TODO |
| /orders | Order.items | Real nested N+1 | Fix with selectinload(Order.items).selectinload(Item.product) (P0) |

**Prioritization:**

1. **P0: /orders** — most hit endpoint, nested N+1 scales with orders × items.
2. **P1: /authors** — simple N+1, easy fix.
3. **P2: /books/{id}** — false positive, whitelist + low priority TODO.

**State after fixes:**
- Tests passing: N of N
- Whitelist: 1 entry (justified)

If your audit finds no N+1: either your app is small/simple and already fine, or nplusone didn't activate correctly. Verify with an endpoint with for x in something: x.relationship that does fire the detector — that confirms the integration works.


Summary and next step

In this capsule you learned:

  • nplusone is the standard detector for N+1 in SQLAlchemy. Modes: raise (tests) and warn (dev).
  • Integration in pytest: NPLUSONE_RAISE=True + NPlusOne(engine) in conftest.py. Any new N+1 breaks CI.
  • Integration in FastAPI: NPlusOne(engine) only if APP_ENV in ("development", "testing"). Never in production.
  • Whitelist for legitimate false positives. Each entry commented with a reason.
  • Custom detector with SQLAlchemy event listeners + ContextVar when nplusone doesn't fit.
  • CI with N+1 detection eliminates an entire class of bugs before they enter production.

Before moving on you should be able to:

  • Configure nplusone raise mode in a pytest project.
  • Configure nplusone warn mode as FastAPI middleware with an env guard.
  • Distinguish a real N+1 from a false positive and decide whitelist vs eager.
  • Integrate the detection into CI with GitHub Actions or an equivalent.

Next capsule — joinedload vs selectinload. You already know how to detect the N+1. Now you learn to solve it. Capsule 04 gives you the two main tools: joinedload (1 query with LEFT JOIN, wins for 1:1 and small 1:N) and selectinload (2 queries with WHERE id IN (...), wins for large 1:N without cartesian explosion). You'll see the exact SQL of each, the plans with EXPLAIN, and the concrete decision matrix for choosing between the two. After capsule 04, when nplusone shouts "n+1 detected" at you, you'll know exactly which option(...) to add to silence it correctly.


Resources

  1. jmcarp/nplusone (GitHub) — the official repo. README with integration examples for SQLAlchemy, Django, Peewee.
  2. SQLAlchemy 2.0 — Events API — for building custom detectors with event.listens_for.
  3. SQLAlchemy 2.0 — before_cursor_execute event — the specific event for counting queries.
  4. pytest-env (PyPI) — plugin for setting env vars in pytest.ini.
  5. GitHub Actions — Postgres service container — canonical pattern for PostgreSQL in CI.
  6. Real Python — "Testing FastAPI with pytest" — fundamentals of async pytest + httpx for FastAPI.
  7. Mike Bayer — "Profile a Python Application" (SQLAlchemy docs) — the performance section of the official FAQ; mentions detection and profiling strategies.

Module 4 — Database Performance & Query Tuning Guide