In-application mode: N+1 detection per request#

1. Problem#

pytest-queryguard only measures inside pytest: a collector is pushed around each test, and class-level listeners on Engine and Session feed every active collector (a process-wide stack). That is right for a test suite, where one test runs at a time, but it cannot run inside a development server:

• concurrent requests (asyncio tasks, worker threads) would all write into the same stack, so one request's statements would be reported for another;
• the listeners are registered on every Engine in the process, not on the ones the application chooses;
• nothing reports per request, and there is nothing to look at while clicking through the application.

2. Goals#

1. Run the same detectors (relationship loads, repeated statements, column loads, optional small repeats and write loops) on every HTTP request of a running ASGI or WSGI application, and report findings through logging, warnings or a callback, with optional per-route query budgets.
2. Statements of concurrent requests, tasks and threads never mix. The one exception is a thread pool on builds where threads inherit the context, used without copy_context().run (see the context-local collector below).
3. Opt in per engine: watch(engine) and unwatch(engine), for Engine and AsyncEngine.
4. No cost when not enabled: importing the module registers nothing.
5. Never break the application: every internal error degrades to one log record and switches the in-app mode off.
6. An optional development panel (JSON and a self-contained HTML page).
7. The pytest plugin keeps its behaviour, its listeners and its tests.

Non-goals: production monitoring, sampling, persistence, authentication of the panel, the missing-index check (it runs EXPLAIN on the request's connection; that belongs in tests), and framework-specific integrations beyond pure ASGI and WSGI.

3. API#

A new subpackage, pytest_queryguard.app, importable without pytest:

from pytest_queryguard import app as queryguard

queryguard.watch(engine)                      # Engine or AsyncEngine
asgi_app = queryguard.QueryGuardASGI(asgi_app, budgets={"GET /orders": 3})
wsgi_app = queryguard.QueryGuardWSGI(flask_app.wsgi_app, report="warn")

with queryguard.track("nightly-job") as report:  # any unit of work
    run_job()

4. Decisions#

• Context-local collector. One ContextVar holds the current RequestReport's collector. asyncio copies the context into each task, Starlette copies it into its thread pool, and SQLAlchemy's async greenlets share the awaiting task's context (gr_context), so a request's tasks and run_in_threadpool calls see its collector while concurrent requests do not. A plain threading.Thread or ThreadPoolExecutor without contextvars.copy_context() starts empty, so its statements are not attributed to any request: missing, never mixed. Where threads inherit the creator's context (Python 3.14 free-threaded builds, or -X thread_inherit_context=1), a pool worker keeps the context of the request that created it, so other requests' tasks can be attributed to that request while it runs. Once that request has finished, the listeners see its tracker as finished and drop the worker's statements (_core.live()). Submitting each task with copy_context().run avoids both on every build; docs/in-app.md says so.

• Separate listeners, per engine. The in-app handlers live in app/_listeners.py and are registered with event.listen(engine, ...) on the watched engine (the sync_engine of an AsyncEngine). The ORM events (do_orm_execute, after_begin) exist only on Session; they are registered on the Session class while at least one engine is watched, and return after one ContextVar.get() outside a request. A relationship or column load is only kept when its statement reaches a watched engine, so sessions on other engines add nothing. The pytest listeners in listeners.py are not changed; the in-app handlers reuse its pure helpers (_is_write, the scope numbers) and the shared QueryCollector, detect() and make_findings().

• Own error state. The pytest listeners stop on their first error through a module global. The in-app mode has its own, so an in-app bug never turns off the measurement of a test run, and the reverse.

• Reporting after the response. Detection runs once, when the request ends (after the last byte for ASGI, after close() of the response iterable for WSGI), never on the statement path.

• Route templates for budgets. Budgets are keyed by "METHOD path" or "path", with fnmatch patterns. Starlette and FastAPI put the matched route in scope["route"], whose path is the template (/orders/{id}). WSGI has no route, so the path is used unless the application calls set_route() (a two-line Flask before_request hook in the docs).

• Panel. Off by default; panel=True serves <panel_path>/ (HTML) and <panel_path>/requests.json from the middleware itself, the last panel_history reports. The HTML has inline CSS and JS only and renders text with textContent. By default it answers loopback clients only; that is a guard against accidental exposure, not access control. Bound parameters are never stored in the panel.

• Memory. max_events (default 1000) caps the statements kept per request; past it they are only counted, as in the plugin.

5. Risks#

• Statements a request starts in a thread without a copied context are not counted (documented).
• Starlette background tasks run inside the request's ASGI call, so their statements count for the request (documented).
• The Session-class listener costs one ContextVar.get() per ORM execution in the whole process while any engine is watched.

Edit this page on GitHub