Counting queries per request to catch N+1 regressions

Increment a ContextVar counter in after_cursor_execute, reset it per request, and assert a budget in your tests — an N+1 then fails in CI instead of degrading production, which no amount of slow-query logging will ever achieve. This is the cheapest instrument in instrumenting and observing async queries and the one most often missing.

Quick Answer

Forty-one fast queries beat one slow one The same endpoint measured two ways. Rendering twenty orders with a lazy relationship issues one query for the orders and one per order for the items — forty-one statements, each taking about two milliseconds. Every one of them is far below any sensible slow-query threshold, so the slow-query log stays empty while the endpoint takes ninety milliseconds of pure round-trip latency. The query count is forty-one against a budget of three, which is unambiguous, actionable and can be asserted in a test. measured by timing 41 statements, ~2 ms each none crosses a 250 ms threshold the slow-query log stays empty the endpoint still takes 90 ms measured by counting 41 statements against a budget of 3 unambiguous and actionable assertable in a test fails in CI, not in production This is why counting belongs first among the instruments: it catches the failure mode that actually degrades applications, and timing structurally cannot see it.

Before — an N+1 nobody can see:

@app.get("/orders")
async def list_orders(session: AsyncSession = Depends(get_session)):
    orders = (await session.execute(select(Order).limit(20))).scalars().all()
    # 20 more queries, one per order, and nothing reports it
    return [{"id": o.id, "items": len(o.items)} for o in orders]

After — the counter, and a test that fails:

from __future__ import annotations

import contextvars
import time
from dataclasses import dataclass, field

from sqlalchemy import event
from sqlalchemy.ext.asyncio import AsyncEngine


@dataclass
class QueryStats:
    count: int = 0
    total_seconds: float = 0.0
    statements: list[str] = field(default_factory=list)


query_stats: contextvars.ContextVar[QueryStats | None] = contextvars.ContextVar(
    "query_stats", default=None
)


def install_query_counting(async_engine: AsyncEngine) -> None:
    engine = async_engine.sync_engine

    @event.listens_for(engine, "before_cursor_execute")
    def _before(conn, cursor, statement, parameters, context, executemany):
        conn.info["query_started"] = time.perf_counter()

    @event.listens_for(engine, "after_cursor_execute")
    def _after(conn, cursor, statement, parameters, context, executemany):
        started = conn.info.pop("query_started", None)
        stats = query_stats.get()
        if stats is None:
            return
        stats.count += 1
        if started is not None:
            stats.total_seconds += time.perf_counter() - started
        if len(stats.statements) < 50:          # bounded, always
            stats.statements.append(" ".join(statement.split())[:200])

# The test that stops it coming back
@pytest.mark.asyncio
async def test_order_list_query_budget(client):
    token = query_stats.set(QueryStats())
    try:
        await client.get("/orders?limit=20")
        stats = query_stats.get()
        assert stats.count <= 3, f"{stats.count} queries:\n" + "\n".join(stats.statements)
    finally:
        query_stats.reset(token)

The assertion message includes the statements, so a failure names the query that fanned out rather than only the number.

Execution Context & Async Workflow Integration

A ContextVar is the right container because asyncio copies the context when a task is created and each task then mutates its own copy. Ten concurrent requests get ten independent counters with no locking, and — importantly — a task that spawns a child task passes its counter down, so queries issued by a gather() inside a handler are counted against that handler.

# Async — middleware that sets, reports and resets
from starlette.middleware.base import BaseHTTPMiddleware


class QueryStatsMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, warn_above: int = 25) -> None:
        super().__init__(app)
        self.warn_above = warn_above

    async def dispatch(self, request, call_next):
        token = query_stats.set(QueryStats())
        try:
            response = await call_next(request)
        finally:
            stats = query_stats.get()
            query_stats.reset(token)        # never leaks into the next request

        response.headers["X-Query-Count"] = str(stats.count)
        response.headers["X-Query-Ms"] = str(round(stats.total_seconds * 1000, 1))
        if stats.count > self.warn_above:
            logger.warning(
                "query fan-out",
                extra={"count": stats.count, "path": request.url.path,
                       "method": request.method},
            )
        return response

The X-Query-Count header is worth shipping even in production. It costs nothing, it makes the cost of an endpoint visible in the browser dev tools, and it turns "this page feels slow" into a number during a support conversation.

One subtlety with background tasks: a BackgroundTask scheduled by a handler runs after the middleware's finally block, in a context copied at task creation. It therefore holds a reference to a QueryStats object nobody is reading any more, and its queries are counted into a discarded object rather than into the request. That is the correct behaviour — the queries genuinely are not part of the request — but it means background work needs its own counter if you want to see it. The lifetime question behind that is the same one in integrating SQLAlchemy async with FastAPI and Starlette: a background task outlives the request that started it.

Per-task counting with a ContextVar Four stages. Middleware sets a fresh stats object into the ContextVar and keeps the returned token. Because a ContextVar is per task under asyncio, ten concurrent requests have ten independent counters with no locking. Every statement the request issues increments whichever counter its own task owns, via the after_cursor_execute listener. At the end the token is reset in a finally block, which is what stops the next request handled by the same task from inheriting the previous count — the same discipline a request-scoped tenant identifier needs. middleware sets a fresh QueryStats and keeps the reset token once per request the ContextVar is per task ten concurrent requests, ten counters no locking needed each statement increments its own from after_cursor_execute whichever task issued it reset(token) in a finally block or the next request inherits the count Skipping the reset produces a bug that only appears under load, when a worker task handles a second request — which is the hardest kind to reproduce deliberately.

Resolving Warnings, Errors & Common Mistakes

SymptomRoot CauseProduction Fix
Counts are wrong only under loadThe token is never reset, so a worker task inherits the previous request's counter.query_stats.reset(token) in a finally block.
Count is always zeroListeners attached to the AsyncEngine rather than engine.sync_engine.Attach to sync_engine.
Count is always zero in testsNo middleware runs in the test, so the ContextVar is None and the listener returns early.Set the ContextVar in the test, as the example does, or provide a fixture that does.
Memory grows during a batch jobThe statements list is unbounded and the job issues millions of queries.Cap the list, as if len(...) < 50 does.
The budget test passes but production fans outFixtures hold one row per collection, so the N+1 does not multiply.Seed enough rows for the fan-out to show, and keep the production warning as a second net.
Counts include queries from other requestsA shared QueryStats object was created once at import rather than per request.Create it inside the middleware, per request.

Advanced: Per-Endpoint Budgets and a Fixture That Enforces Them

A single global budget is a blunt instrument: a list endpoint legitimately issues more queries than a health check. Declaring the budget next to the route makes it reviewable, and a fixture can enforce it without every test remembering to.

# routes.py — the budget lives beside the handler it describes
from myapp.observability import query_budget


@app.get("/orders")
@query_budget(3)          # one for orders, one for items, one for the count
async def list_orders(session: AsyncSession = Depends(get_session)):
    ...

# observability.py — the decorator records the budget on the endpoint
from typing import Callable, TypeVar

F = TypeVar("F", bound=Callable)


def query_budget(maximum: int) -> Callable[[F], F]:
    def decorate(func: F) -> F:
        func.__query_budget__ = maximum      # read by tests and by middleware
        return func
    return decorate

# conftest.py — a fixture that checks the budget for every request a test makes
@pytest.fixture(autouse=True)
def enforce_query_budgets(request, monkeypatch):
    """Fail any test whose requests exceed the budget declared on the endpoint."""
    breaches: list[str] = []

    original = QueryStatsMiddleware.dispatch

    async def checked(self, http_request, call_next):
        token = query_stats.set(QueryStats())
        try:
            response = await call_next(http_request)
        finally:
            stats = query_stats.get()
            query_stats.reset(token)
        endpoint = http_request.scope.get("endpoint")
        budget = getattr(endpoint, "__query_budget__", None)
        if budget is not None and stats.count > budget:
            breaches.append(
                f"{http_request.url.path}: {stats.count} queries, budget {budget}\n  "
                + "\n  ".join(stats.statements)
            )
        return response

    monkeypatch.setattr(QueryStatsMiddleware, "dispatch", checked)
    yield
    assert not breaches, "query budget exceeded:\n" + "\n".join(breaches)

Two properties make this worth the setup. It is autouse, so a new endpoint test is covered without anyone remembering — and a new endpoint with no declared budget is simply not checked, which keeps the mechanism from blocking work. And the failure message carries the statements, so the diagnosis is in the test output rather than requiring a re-run with logging enabled.

Seeding matters as much as the mechanism. A fixture that creates one order with one item cannot produce a fan-out, so the budget test passes on code that will fan out in production. Seed enough rows that N is visibly greater than one — five is usually sufficient — and the test means what it claims.

Three budgets, three different jobs Three uses of the same number. A hard budget asserted in a test fails the build when a loader option is removed, which is the only reliable way to keep an N+1 fixed once it is found. A soft budget in production logs a warning above a threshold, catching handlers that fan out only against real data volumes rather than against test fixtures. And a per-endpoint budget recorded next to the route makes the expected cost explicit, so a reviewer can see when a change to a handler ought to have changed its budget too. hard budget, in tests assert count <= N fails the build stops regressions soft budget, in production warn above a threshold catches real data volumes fixtures are too small to see it per-endpoint budget declared next to the route visible in review a change to it is a change to review The test budget is the one that pays for itself immediately; the production warning is what finds the handlers whose fan-out depends on how much data a customer actually has.

Reading the Numbers: What a Healthy Count Looks Like

Once counts are visible, the useful question is what to expect, and a few rules of thumb hold up across most applications.

A read endpoint returning a list should issue one query for the page, one per eagerly-loaded collection, and at most one for a total count. Three to five is normal; twenty is a fan-out. If the number grows with the page size, it is an N+1 by definition — that is the diagnostic, and it does not need a profiler.

A write endpoint should issue the reads it needs, then a small number of writes. What inflates it is usually per-row work in a loop: a validation that re-queries, a session.flush() inside an iteration, or a relationship access on each new object. Batching those is the same reasoning as in high-performance bulk inserts and updates, applied at a smaller scale.

A health check should issue zero or one. A health check that issues four is checking things it should not, and it will be the first thing to fail when the database is under stress — reporting the service unhealthy when it is merely busy.

The number worth watching over time is not the absolute count but its ratio to page size. A constant count as the page grows means the loaders are right. A count that tracks the page size means at least one relationship is loading lazily, whatever the absolute figure happens to be today.

Frequently Asked Questions

Why a ContextVar rather than a task-local or a thread-local?

A thread-local is wrong under asyncio because many tasks share one thread. A ContextVar is copied when a task is created and mutated independently thereafter, which gives exactly per-request isolation, and it propagates into child tasks so a gather() inside a handler is still counted against that handler.

Does the counter add measurable overhead?

No. It is a ContextVar lookup, an integer increment and a bounded list append per statement — nanoseconds against a query measured in milliseconds. It is cheap enough to leave enabled in production, which is where it catches the fan-outs that only appear at real data volumes.

Should the budget be a hard failure in production?

No — warn, do not fail. A count over budget means an endpoint is inefficient, not that the response is wrong, and refusing to serve it makes a performance problem into an outage. Fail hard in tests, warn in production.

How do I count queries in a background job rather than a request?

Set the ContextVar at the top of the job and reset it at the end, exactly as the middleware does. For a job processing batches, setting it per batch gives a per-batch count, which is usually the more useful number than a total that only grows.