queryargus 0.1.1

Scope-aware SQL query tracing for Python applications, tests, and offline reports.

QueryArgus

See every query. Miss nothing.

QueryArgus is a Python library for tracing SQL queries across requests, jobs, tests, CLI commands, and regular functions. It captures query execution with business context, stores finished traces in simple offline-friendly formats, and helps you understand repeated queries, hotspots, and possible N+1 sequences without forcing an APM-shaped workflow on your application.

Portuguese (Brazil) documentation

Why QueryArgus

Database behavior often becomes opaque as an application grows:

• N+1 issues slip into production unnoticed.
• Performance regressions show up late.
• Generic APMs reveal symptoms, but not the actual query flow inside the unit of work you care about.
• Plain logs rarely connect a query to the request, job, or repository method that caused it.

QueryArgus exists to make query behavior visible, understandable, and actionable.

What it does

• Captures SQL queries through pluggable adapters.
• Associates queries with a root trace and nested business scopes.
• Infers a useful scope from caller code when you do not set one explicitly.
• Works in HTTP apps, background jobs, worker handlers, tests, and CLI code.
• Persists finished traces to JSONL or memory sinks.
• Generates offline reports in text, JSON, HTML, SVG badge, and Markdown summary formats.
• Provides pytest fixtures and assertions for query-aware tests.

Installation

Install the core package:

pip install queryargus

Install common integrations:

pip install "queryargus[sqlalchemy,fastapi,testing]"

Available extras:

• sqlalchemy: SQLAlchemy Engine and AsyncEngine instrumentation.
• psycopg: psycopg 3 sync and async connection instrumentation.
• asyncpg: asyncpg connection instrumentation.
• fastapi: FastAPI and Starlette middleware integration.
• testing: pytest plugin and testing helpers.

Quick start

setup() is the recommended entrypoint when you want one line to configure sinks, database instrumentation, and framework integration:

from fastapi import FastAPI
from sqlalchemy import create_engine
import queryargus

app = FastAPI()
engine = create_engine("sqlite:///app.db")

queryargus.setup(
    adapter=engine,
    framework=app,
    sink="jsonl",
)

With that in place:

• the SQLAlchemy adapter captures queries automatically
• the FastAPI middleware opens one trace per request
• traces are appended to traces/queryargus.jsonl

Core usage patterns

FastAPI or Starlette request tracing

from fastapi import FastAPI
from sqlalchemy import create_engine
import queryargus

app = FastAPI()
engine = create_engine("sqlite:///inventory.db")

queryargus.setup(adapter=engine, framework=app, sink="jsonl")

Each HTTP request becomes a root trace named like GET /products.

Jobs, workers, and CLI functions

Use @traced when the function itself is the unit of work:

import queryargus

queryargus.setup(adapter=engine, sink="jsonl")

@queryargus.traced("billing.process_invoice")
def process_invoice(invoice_id: str) -> None:
    repository.load_invoice(invoice_id)
    service.apply_rules(invoice_id)
    repository.mark_processed(invoice_id)

If a trace is already active, @traced does not open a nested root trace.

Manual tracing

Use start_trace() when a context manager fits better than a decorator:

from queryargus import start_trace, trace_scope

with start_trace("inventory.rebuild_projection"):
    with trace_scope("inventory_repository.load_snapshot"):
        repository.load_snapshot()

Important behavior:

• start_trace() raises NestedTraceError if a root trace is already active.
• trace_scope() is a no-op when there is no active trace.

Automatic business scopes on repositories and services

Instrument a class:

from queryargus import trace_methods

@trace_methods()
class StockRepository:
    def get_by_id(self, stock_id: str):
        ...

    def list_all(self):
        ...

Instrument an existing instance:

from queryargus import instrument_object_methods

repository = instrument_object_methods(repository, namespace="products")

That gives you scope labels such as StockRepository.get_by_id or products.list_all in reports.

Supported adapters and integrations

SQLAlchemy

Auto-detected by setup(adapter=engine) or explicitly installed:

from queryargus.adapters import instrument_sqlalchemy

instrument_sqlalchemy(engine)

psycopg 3

from queryargus.adapters import instrument_psycopg

conn = instrument_psycopg(conn)

asyncpg

from queryargus.adapters import instrument_asyncpg

conn = instrument_asyncpg(conn)

FastAPI and Starlette

Auto-detected by setup(framework=app) or explicitly installed:

from queryargus.integrations.fastapi import instrument_fastapi
from queryargus.integrations.starlette import instrument_starlette

instrument_fastapi(app)
instrument_starlette(app)

Sinks

QueryArgus ships with two built-in sinks:

• JsonlSink: appends one serialized trace per line to a JSONL file
• MemorySink: keeps traces in memory for tests and local experiments

Examples:

import queryargus

queryargus.setup(sink="jsonl")
queryargus.setup(sink="memory")
queryargus.setup(sink=[queryargus.MemorySink(), queryargus.JsonlSink("var/traces.jsonl")])

When you omit sink, QueryArgus defaults to JsonlSink("traces/queryargus.jsonl").

Testing support

Install the testing extra:

pip install "queryargus[testing]"

Pytest fixtures exposed through the pytest11 entry point:

• queryargus_memory_sink
• queryargus_trace
• queryargus_captured

Useful helpers:

from queryargus.testing import (
    assert_all_queries_scoped,
    assert_max_duration_ms,
    assert_no_n_plus_one,
    assert_no_repeated_queries,
    assert_query_count_at_most,
    capture_queries,
)

Example:

def test_repository_is_efficient(queryargus_trace):
    repository.list_products()

    assert_query_count_at_most(queryargus_trace, 3)
    assert_no_n_plus_one(queryargus_trace)

Offline reports

Generate an offline report from collected traces:

queryargus-report PATH [--format text|json|html|badge|summary] [--output OUTPUT]

Examples:

queryargus-report traces/queryargus.jsonl --format html
queryargus-report traces/queryargus.jsonl --format summary --output artifacts/queryargus-summary.md
queryargus-report traces/queryargus.jsonl --format badge --output artifacts/queryargus-coverage.svg
queryargus-report traces/queryargus.jsonl --format badge --badge-type n-plus-one --output artifacts/queryargus-n-plus-one.svg

Analytic Visions (--format):

• html: Analytic View: Full interactive report for deep navigation, hotspot analysis, and N+1. (Recommended for local use)
• summary: Code Review View: Markdown summary ideal for Pull Request comments or CI logs.
• badge: Status View: Visual badge for displaying coverage percentage in READMEs. Use --badge-type n-plus-one to generate an N+1 badge.
• text: Terminal View: Quick summary for immediate check in the console.
• json: Data View: Full structured payload for integrations and automation.

Output behavior:

• <input_stem>_queryargus_report
• For html, --output is a directory.
• For summary and badge, --output is a file path.

The HTML report includes coverage summaries, entrypoint breakdowns, hotspot analysis, and N+1 detection.

Project files

• English guide: README.md
• Brazilian Portuguese guide: README.pt-BR.md
• Contribution guide: CONTRIBUTING.md
• Change history: CHANGELOG.md

Compatibility

• Python >=3.10,<3.15
• Typed package (py.typed included)
• Offline-first JSONL workflow by default
• Designed for application code, tests, and CI pipelines

Development status

The package is ready for early professional usage and packaging, with a deliberately small public API. The current release focuses on reliable trace capture, scope-aware analysis, and offline reporting.