n-plus-one-sniffer

Catch N+1 query patterns in SQLAlchemy 2.0 code — in your tests and at runtime — and get told exactly how to fix them.

The N+1 problem is the most common ORM performance bug: you load a list of parents, then touch a relationship in a loop, and the ORM quietly fires one extra query per row. It never crashes. It just gets slower as your data grows, and it slips through code review because the offending line looks innocent.

n-plus-one-sniffer watches the queries your code actually runs, collapses each SQL statement to a parameter-agnostic fingerprint, and flags any fingerprint that repeats more than a threshold inside a monitored block. It points at the line of your code that triggered the queries and recommends the right eager-loading strategy (selectinload, joinedload, or subqueryload) with the reasoning spelled out.

---

The problem, made visible

Here is a textbook N+1 — one query for the users, then one more for every user's addresses:

# ❌ Before: 1 + N queries
with Session(engine) as session:
    users = session.scalars(select(User)).all()   # 1 query
    for user in users:
        print(user.name, len(user.addresses))     # +1 query each, silently

Run it under the sniffer and the problem stops being invisible:

from n_plus_one_sniffer import catch_n_plus_one

with catch_n_plus_one(engine, threshold=3) as sniffer:
    with Session(engine) as session:
        users = session.scalars(select(User)).all()
        for user in users:
            _ = len(user.addresses)

print(sniffer.report())

Detected 1 suspected N+1 query pattern(s) (threshold=3):

N+1 suspected: 50 executions (threshold 3) of a query on 'addresses'
  origin:   /app/reports.py:42 in build_report
  sql:      SELECT addresses.id, addresses.email, addresses.user_id FROM addresses WHERE addresses.user_id = ?
  fix:      likely selectinload
  why:      This looks like a per-parent foreign-key lookup (SELECT ... WHERE fk = ?)
            repeated once per parent row. That is the classic collection N+1. Use
            selectinload(Parent.relationship): it emits a single extra
            SELECT ... WHERE fk IN (...) batched across all parents ...

The fix it points you to:

# ✅ After: 2 queries total, regardless of how many users
from sqlalchemy.orm import selectinload

with Session(engine) as session:
    stmt = select(User).options(selectinload(User.addresses))
    users = session.scalars(stmt).all()            # 1 query
    for user in users:                             # addresses already loaded
        print(user.name, len(user.addresses))      # 0 extra queries

---

Install

Not on PyPI. Install straight from GitHub:

pip install "git+https://github.com/async-workflows/n-plus-one-sniffer.git"

Requires Python 3.10+ and SQLAlchemy 2.0+.

---

Usage: the context manager

catch_n_plus_one(engine, threshold=3) returns a sniffer you use as a context manager. It works with a sync Engine and with an AsyncEngine (it hooks the underlying sync_engine).

from n_plus_one_sniffer import catch_n_plus_one, NPlusOneError

with catch_n_plus_one(engine, threshold=3) as sniffer:
    run_the_code_under_test()

# Structured findings:
for v in sniffer.violations:
    print(v.table, v.count, v.strategy, v.origin)

# Or a formatted report:
print(sniffer.report())

Each item in sniffer.violations is a Violation with fingerprint, sql, count, threshold, table, strategy, explanation, and origin (the file:line in function of your code that triggered it).

Pass raise_on_violation=True to turn any detection into a raised NPlusOneError:

with catch_n_plus_one(engine, threshold=3, raise_on_violation=True):
    run_the_code_under_test()   # raises NPlusOneError if an N+1 is found

---

Usage: the pytest plugin

The plugin is registered automatically on install (via the pytest11 entry point). Bind it to your engine by defining a fixture named nplusone_engine that returns the engine you want watched:

# conftest.py
import pytest
from myapp.db import engine   # your sync Engine or AsyncEngine

@pytest.fixture
def nplusone_engine():
    return engine

The no_n_plus_one marker

Mark a test and it fails automatically if an N+1 pattern happens while it runs:

import pytest

@pytest.mark.no_n_plus_one
def test_report_has_no_n_plus_one(nplusone_engine):
    build_report()   # fails the test if this fires N+1 queries

@pytest.mark.no_n_plus_one(threshold=5)   # per-test override
def test_tolerates_a_few(nplusone_engine):
    ...

The n_plus_one fixture

Request the fixture when you want to assert on findings yourself instead of just failing:

def test_inspect(n_plus_one, nplusone_engine):
    build_report()
    assert not n_plus_one_violations_yet(n_plus_one)   # your own assertions
    # n_plus_one.violations / n_plus_one.report() are available after the block

The n_plus_one fixture reflects only the queries that have run so far when you read .violations; the sniffer finalises when the fixture's block ends. For a hard pass/fail gate, prefer the marker.

Threshold configuration

Set the default duplicate-query threshold on the CLI or in your ini file:

pytest --n-plus-one-threshold=5

# pytest.ini / pyproject.toml [tool.pytest.ini_options]
n_plus_one_threshold = 5

Precedence: @pytest.mark.no_n_plus_one(threshold=...) > --n-plus-one-threshold > ini n_plus_one_threshold > built-in default (3).

---

Usage: at runtime in a web app

The context manager is the reusable runtime primitive — wrap a unit of work (a request) with it. Here it is as a FastAPI dependency that logs a warning when a request triggers an N+1:

import logging
from fastapi import FastAPI, Depends
from n_plus_one_sniffer import catch_n_plus_one
from myapp.db import engine

log = logging.getLogger("n_plus_one")
app = FastAPI()

def sniff_request():
    with catch_n_plus_one(engine, threshold=10) as sniffer:
        yield                      # the request handler runs here
    for v in sniffer.violations:   # inspected after the response is built
        log.warning("N+1 in request: %s", v)

@app.get("/users", dependencies=[Depends(sniff_request)])
def list_users():
    ...

The same shape works as ASGI-style middleware: open the sniffer before calling the app, read sniffer.violations after. Use a higher threshold in production than in tests — a busy request legitimately issues more queries than a focused unit test.

---

How detection and recommendations work

• Listener. On entering the block, the sniffer registers a SQLAlchemy before_cursor_execute event listener on the engine (the sync_engine for an AsyncEngine). It removes the listener on exit, so nothing leaks between blocks.
• Fingerprinting. Every statement is normalised: string and numeric literals and bound-parameter placeholders (?, %s, %(name)s, :name, $1) become ?, variable-length IN (?, ?, ?) lists collapse to IN (?), and whitespace is squeezed. So the same template run 50 times with different parameters is one fingerprint with 50 executions.
• Flagging. Any fingerprint that executes more than threshold times inside the block is reported. The sniffer also captures a short application stack frame — skipping SQLAlchemy, site-packages, and its own frames — so the report points at your code.
• Recommendation (honest heuristics). A repeated simple SELECT ... WHERE fk = ? with no join → likely selectinload (batches into one IN (...) query, and won't multiply rows on a one-to-many). A repeated query with a JOIN, which usually means a lazily-loaded to-one → joinedload (folds it into one SELECT), with subqueryload mentioned as an alternative for large batches. The report says "likely" and shows its reasoning — treat it as a strong hint, not gospel.

---

Tuning false positives

• Raise the threshold. Some duplicate queries are legitimate. The threshold is the whole tuning knob: a small helper called a handful of times inside a request is normal; the same helper called once per row of a 10,000-row result set is not. Set it per test with the marker, globally with --n-plus-one-threshold, or per call with catch_n_plus_one(engine, threshold=...).
• Scope the block tightly. Only queries inside the with block are counted. Wrap the specific operation you care about rather than an entire test, so unrelated fixtures and setup don't inflate counts.
• Batched eager loads are fine. selectinload itself may emit its IN (...) query in chunks for very large parent sets — that is a handful of executions, not N, and normally stays under a sensible threshold.

Limitations

• Detection is statistical, not semantic: it sees that a fingerprint repeated, not why. A genuinely necessary loop of identical queries will look like an N+1 — that is what the threshold is for.
• Recommendations are heuristic hints based on SQL shape, not a full query-plan analysis. Verify the suggested loading strategy against your relationship's cardinality.
• Only queries that reach the DBAPI cursor on the monitored engine are seen. Results served entirely from SQLAlchemy's identity map (already-loaded objects) issue no SQL and are correctly not flagged.
• Fingerprinting is tuned for standard parameterised SQL. Statements built with inlined literals via unusual string formatting may fingerprint less tightly.

---

Further reading

A few pointers if you want to go deeper on the fixes this tool recommends: the broader advanced query patterns and bulk data operations guide, the specifics of relationship loading strategies, a focused comparison of selectinload vs joinedload for N+1 prevention, and how to avoid cartesian product warnings in SQLAlchemy joins.

License

MIT © 2026 async-workflows