Skip to content

API Reference

Auto-generated documentation

This page uses mkdocstrings to render API documentation directly from source code docstrings. Some references may not render if module paths differ from the documented structure.

Core

QueryDoctorMiddleware

The Django middleware that intercepts queries for each request and runs them through the analysis pipeline.

QueryDoctorMiddleware(get_response)

Django middleware that activates query diagnosis per request.

Installs an execute_wrapper on the database connection to capture all SQL queries. After the response is generated, runs all enabled analyzers and sends reports to configured reporters.

Supports both sync and async views via sync_capable and async_capable.

Initialize the middleware.

Parameters:

Name Type Description Default
get_response Callable[..., Any]

The next middleware or view in the chain.

 required
Source code in src/query_doctor/middleware.py 
77
78
79
80
81
82
83
84
def __init__(self, get_response: Callable[..., Any]) -> None:
    """Initialize the middleware.

    Args:
        get_response: The next middleware or view in the chain.
    """
    self.get_response = get_response
    self._is_async = inspect.iscoroutinefunction(get_response)
__acall__(request) async

Process an async request through the query doctor pipeline.

Parameters:

Name Type Description Default
request HttpRequest

The incoming HTTP request.

 required

Returns:

Type Description
Any

The HTTP response from the async view.
Source code in src/query_doctor/middleware.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
async def __acall__(self, request: HttpRequest) -> Any:
    """Process an async request through the query doctor pipeline.

    Args:
        request: The incoming HTTP request.

    Returns:
        The HTTP response from the async view.
    """
    try:
        config = get_config()
    except Exception:
        logger.warning("query_doctor: failed to load config", exc_info=True)
        return await self.get_response(request)

    if not self._should_process(request, config):
        return await self.get_response(request)

    interceptor = QueryInterceptor(capture_stack=config.get("CAPTURE_STACK_TRACES", True))

    from django.db import connection

    with connection.execute_wrapper(interceptor):
        response = await self.get_response(request)

    try:
        self._analyze_and_report(interceptor, config, request)
    except Exception:
        logger.warning("query_doctor: analysis failed", exc_info=True)

    return response
__call__(request)

Process a request through the query doctor pipeline.

Routes to sync or async path based on the get_response type.

Parameters:

Name Type Description Default
request HttpRequest

The incoming HTTP request.

 required

Returns:

Type Description
Any

The HTTP response from the view.
Source code in src/query_doctor/middleware.py 
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def __call__(self, request: HttpRequest) -> Any:
    """Process a request through the query doctor pipeline.

    Routes to sync or async path based on the get_response type.

    Args:
        request: The incoming HTTP request.

    Returns:
        The HTTP response from the view.
    """
    if self._is_async:
        return self.__acall__(request)
    return self._sync_call(request)

Context Managers

Context managers for targeted query analysis outside of the middleware.

context_managers

Context managers for targeted query diagnosis.

Provides diagnose_queries() for diagnosing queries within a specific code block rather than an entire request.

diagnose_queries()

Context manager for targeted query diagnosis.

Captures and analyzes all SQL queries executed within the context. The DiagnosisReport is yielded and populated after the context exits.

Usage

with diagnose_queries() as report: # ... your ORM code here ... print(report.issues)

Source code in src/query_doctor/context_managers.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
@contextmanager
def diagnose_queries() -> Generator[DiagnosisReport, None, None]:
    """Context manager for targeted query diagnosis.

    Captures and analyzes all SQL queries executed within the context.
    The DiagnosisReport is yielded and populated after the context exits.

    Usage:
        with diagnose_queries() as report:
            # ... your ORM code here ...
        print(report.issues)
    """
    report = DiagnosisReport()
    interceptor = QueryInterceptor()

    from django.db import connection

    with connection.execute_wrapper(interceptor):
        yield report

    # After context exits, run analysis
    try:
        queries = interceptor.get_queries()
        report.captured_queries = queries
        report.total_queries = len(queries)
        report.total_time_ms = sum(q.duration_ms for q in queries)

        # Run analyzers
        analyzers = [NPlusOneAnalyzer(), DuplicateAnalyzer(), MissingIndexAnalyzer()]
        for analyzer in analyzers:
            try:
                prescriptions = analyzer.analyze(queries)
                report.prescriptions.extend(prescriptions)
            except Exception:
                logger.warning(
                    "query_doctor: analyzer %s failed in context manager",
                    analyzer.name,
                    exc_info=True,
                )
    except Exception:
        logger.warning("query_doctor: context manager analysis failed", exc_info=True)

Decorators

Function and method decorators for query diagnosis and budgets.

decorators

Decorators for query diagnosis and budget enforcement.

Provides @diagnose for wrapping functions with automatic query analysis, and @query_budget for enforcing query count and time limits.

diagnose(func)

Decorator that diagnoses queries executed within a function.

Wraps the function with diagnose_queries() context manager. After execution, the DiagnosisReport is attached as func._query_doctor_report.

Usage

@diagnose def my_view(request): return Book.objects.all()

Source code in src/query_doctor/decorators.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def diagnose(func: Any) -> Any:
    """Decorator that diagnoses queries executed within a function.

    Wraps the function with diagnose_queries() context manager. After
    execution, the DiagnosisReport is attached as func._query_doctor_report.

    Usage:
        @diagnose
        def my_view(request):
            return Book.objects.all()
    """

    @functools.wraps(func)
    def wrapper(*args: Any, **kwargs: Any) -> Any:
        """Execute the wrapped function with query diagnosis."""
        try:
            with diagnose_queries() as report:
                result = func(*args, **kwargs)
        except QueryBudgetError:
            raise
        except Exception:
            # If diagnose_queries itself fails to set up, run without diagnosis
            logger.warning(
                "query_doctor: @diagnose failed, running function without diagnosis",
                exc_info=True,
            )
            return func(*args, **kwargs)

        wrapper._query_doctor_report = report  # type: ignore[attr-defined]
        return result

    return wrapper
query_budget(max_queries=None, max_time_ms=None)

Decorator that enforces query budget limits on a function.

Raises QueryBudgetError if the function exceeds the specified query count or time limits. Falls back to config defaults if no explicit limits are provided.

Parameters:

Name Type Description Default
max_queries int | None

Maximum number of queries allowed. None means no limit.

 None
max_time_ms float | None

Maximum total query time in milliseconds. None means no limit.

 None
Usage

@query_budget(max_queries=10, max_time_ms=100) def my_view(request): return Book.objects.all()

Source code in src/query_doctor/decorators.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def query_budget(
    max_queries: int | None = None,
    max_time_ms: float | None = None,
) -> Any:
    """Decorator that enforces query budget limits on a function.

    Raises QueryBudgetError if the function exceeds the specified
    query count or time limits. Falls back to config defaults if no
    explicit limits are provided.

    Args:
        max_queries: Maximum number of queries allowed. None means no limit.
        max_time_ms: Maximum total query time in milliseconds. None means no limit.

    Usage:
        @query_budget(max_queries=10, max_time_ms=100)
        def my_view(request):
            return Book.objects.all()
    """

    def decorator(func: Any) -> Any:
        """Wrap the function with budget enforcement."""

        @functools.wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            """Execute the wrapped function with budget checking."""
            # Resolve limits: explicit args take priority over config
            config = get_config()
            budget_config = config.get("QUERY_BUDGET", {})
            effective_max_queries = (
                max_queries
                if max_queries is not None
                else budget_config.get("DEFAULT_MAX_QUERIES")
            )
            effective_max_time_ms = (
                max_time_ms
                if max_time_ms is not None
                else budget_config.get("DEFAULT_MAX_TIME_MS")
            )

            with diagnose_queries() as report:
                result = func(*args, **kwargs)

            wrapper._query_doctor_report = report  # type: ignore[attr-defined]

            # Check budget after execution
            if effective_max_queries is not None and report.total_queries > effective_max_queries:
                raise QueryBudgetError(
                    f"Query budget exceeded: {report.total_queries} queries "
                    f"(max_queries={effective_max_queries})",
                    report=report,
                )

            report_time = _get_report_time_ms(report)
            if effective_max_time_ms is not None and report_time > effective_max_time_ms:
                raise QueryBudgetError(
                    f"Query budget exceeded: {report_time:.1f}ms "
                    f"(max_time_ms={effective_max_time_ms})",
                    report=report,
                )

            return result

        return wrapper

    return decorator

Data Types

Severity

Severity levels for prescriptions.

Severity

Bases: Enum

Severity level for a diagnosed issue.

Prescription

The data class returned by analyzers describing a detected issue and its fix.

Prescription(issue_type, severity, description, fix_suggestion, callsite, query_count=0, time_saved_ms=0, fingerprint='', extra=dict()) dataclass

A diagnosed issue with an actionable fix.

CapturedQuery

Information captured for each SQL query during interception.

CapturedQuery(sql, params, duration_ms, fingerprint, normalized_sql, callsite, is_select, tables) dataclass

A single SQL query captured during a request.

DiagnosisReport

Aggregated report of all prescriptions for a request or command.

DiagnosisReport(prescriptions=list(), total_queries=0, total_time_ms=0, captured_queries=list()) dataclass

Complete report for one request/context.

has_critical property

Return True if any prescription has CRITICAL severity.

issues property

Return the total number of diagnosed issues.

n_plus_one_count property

Return the count of N+1 query issues.

Analyzers

BaseAnalyzer

The abstract base class that all analyzers implement. Subclass this to create custom analyzers.

BaseAnalyzer

Bases: ABC

Base class for all query analyzers.

analyze(queries, models_meta=None) abstractmethod

Analyze captured queries and return prescriptions.

Parameters:

Name Type Description Default
queries list[CapturedQuery]

List of captured SQL queries to analyze.

 required
models_meta dict[str, Any] | None

Optional Django model metadata for enhanced analysis.

 None

Returns:

Type Description
list[Prescription]

List of Prescription objects describing detected issues and fixes.
Source code in src/query_doctor/analyzers/base.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
@abstractmethod
def analyze(
    self,
    queries: list[CapturedQuery],
    models_meta: dict[str, Any] | None = None,
) -> list[Prescription]:
    """Analyze captured queries and return prescriptions.

    Args:
        queries: List of captured SQL queries to analyze.
        models_meta: Optional Django model metadata for enhanced analysis.

    Returns:
        List of Prescription objects describing detected issues and fixes.
    """
    ...
is_enabled()

Check if this analyzer is enabled in config.

Source code in src/query_doctor/analyzers/base.py 
37
38
39
40
41
42
43
def is_enabled(self) -> bool:
    """Check if this analyzer is enabled in config."""
    from query_doctor.conf import get_config

    config = get_config()
    analyzer_config = config.get("ANALYZERS", {}).get(self.name, {})
    return bool(analyzer_config.get("enabled", True))

NPlusOneAnalyzer

Detects N+1 query patterns using fingerprint-based grouping.

NPlusOneAnalyzer

Bases: BaseAnalyzer

Analyzer that detects N+1 query patterns.

Groups queries by fingerprint and identifies repeated SELECT queries that indicate missing select_related() or prefetch_related() calls.

analyze(queries, models_meta=None)

Analyze queries for N+1 patterns.

Parameters:

Name Type Description Default
queries list[CapturedQuery]

List of captured queries to analyze.

 required
models_meta dict[str, Any] | None

Optional model metadata (not used currently).

 None

Returns:

Type Description
list[Prescription]

List of prescriptions for detected N+1 issues.
Source code in src/query_doctor/analyzers/nplusone.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def analyze(
    self,
    queries: list[CapturedQuery],
    models_meta: dict[str, Any] | None = None,
) -> list[Prescription]:
    """Analyze queries for N+1 patterns.

    Args:
        queries: List of captured queries to analyze.
        models_meta: Optional model metadata (not used currently).

    Returns:
        List of prescriptions for detected N+1 issues.
    """
    if not queries:
        return []

    try:
        return self._detect_nplusone(queries)
    except Exception:
        logger.warning("query_doctor: N+1 analysis failed", exc_info=True)
        return []

DuplicateAnalyzer

Detects exact and near-duplicate queries within a single request.

DuplicateAnalyzer

Bases: BaseAnalyzer

Analyzer that detects exact and near-duplicate queries.

Exact duplicates: same SQL + same params executed multiple times. Near-duplicates: same SQL structure (fingerprint) with different params.

analyze(queries, models_meta=None)

Analyze queries for duplicate patterns.

Parameters:

Name Type Description Default
queries list[CapturedQuery]

List of captured queries to analyze.

 required
models_meta dict[str, Any] | None

Optional model metadata (not used).

 None

Returns:

Type Description
list[Prescription]

List of prescriptions for detected duplicate issues.
Source code in src/query_doctor/analyzers/duplicate.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def analyze(
    self,
    queries: list[CapturedQuery],
    models_meta: dict[str, Any] | None = None,
) -> list[Prescription]:
    """Analyze queries for duplicate patterns.

    Args:
        queries: List of captured queries to analyze.
        models_meta: Optional model metadata (not used).

    Returns:
        List of prescriptions for detected duplicate issues.
    """
    if not queries:
        return []

    try:
        return self._detect_duplicates(queries)
    except Exception:
        logger.warning("query_doctor: duplicate analysis failed", exc_info=True)
        return []

MissingIndexAnalyzer

Detects queries filtering or ordering on non-indexed columns.

MissingIndexAnalyzer

Bases: BaseAnalyzer

Analyzer that detects queries on non-indexed columns.

Examines WHERE and ORDER BY clauses to find columns that lack database indexes, and suggests adding Meta.indexes with models.Index().

analyze(queries, models_meta=None)

Analyze queries for missing index issues.

Parameters:

Name Type Description Default
queries list[CapturedQuery]

List of captured queries to analyze.

 required
models_meta dict[str, Any] | None

Optional model metadata (not used).

 None

Returns:

Type Description
list[Prescription]

List of prescriptions for detected missing index issues.
Source code in src/query_doctor/analyzers/missing_index.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def analyze(
    self,
    queries: list[CapturedQuery],
    models_meta: dict[str, Any] | None = None,
) -> list[Prescription]:
    """Analyze queries for missing index issues.

    Args:
        queries: List of captured queries to analyze.
        models_meta: Optional model metadata (not used).

    Returns:
        List of prescriptions for detected missing index issues.
    """
    if not queries:
        return []

    try:
        return self._detect_missing_indexes(queries)
    except Exception:
        logger.warning("query_doctor: missing index analysis failed", exc_info=True)
        return []

FatSelectAnalyzer

Detects queries selecting more columns than necessary.

FatSelectAnalyzer(field_count_threshold=None)

Bases: BaseAnalyzer

Analyzer that detects overly broad SELECT queries.

Flags queries that select many columns and suggests using .only() or .defer() to reduce data transfer.

Initialize the analyzer.

Parameters:

Name Type Description Default
field_count_threshold int | None

Minimum number of columns to flag. Defaults to config or 8.

 None
Source code in src/query_doctor/analyzers/fat_select.py 
57
58
59
60
61
62
63
64
def __init__(self, field_count_threshold: int | None = None) -> None:
    """Initialize the analyzer.

    Args:
        field_count_threshold: Minimum number of columns to flag.
                              Defaults to config or 8.
    """
    self._threshold_override = field_count_threshold
analyze(queries, models_meta=None)

Analyze queries for fat SELECT patterns.

Parameters:

Name Type Description Default
queries list[CapturedQuery]

List of captured queries to analyze.

 required
models_meta dict[str, Any] | None

Optional model metadata (not used currently).

 None

Returns:

Type Description
list[Prescription]

List of prescriptions for detected fat SELECT issues.
Source code in src/query_doctor/analyzers/fat_select.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def analyze(
    self,
    queries: list[CapturedQuery],
    models_meta: dict[str, Any] | None = None,
) -> list[Prescription]:
    """Analyze queries for fat SELECT patterns.

    Args:
        queries: List of captured queries to analyze.
        models_meta: Optional model metadata (not used currently).

    Returns:
        List of prescriptions for detected fat SELECT issues.
    """
    if not queries or not self.is_enabled():
        return []

    try:
        return self._detect_fat_selects(queries)
    except Exception:
        logger.warning("query_doctor: fat SELECT analysis failed", exc_info=True)
        return []

QuerySetEvalAnalyzer

Detects unintended queryset evaluation patterns.

QuerySetEvalAnalyzer

Bases: BaseAnalyzer

Analyzer that detects inefficient queryset evaluation patterns.

Inspects call site code context to find patterns where Django provides more efficient alternatives (count, exists, first).

analyze(queries, models_meta=None)

Analyze queries for inefficient evaluation patterns.

Parameters:

Name Type Description Default
queries list[CapturedQuery]

List of captured queries to analyze.

 required
models_meta dict[str, Any] | None

Optional model metadata (not used).

 None

Returns:

Type Description
list[Prescription]

List of prescriptions for detected evaluation issues.
Source code in src/query_doctor/analyzers/queryset_eval.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def analyze(
    self,
    queries: list[CapturedQuery],
    models_meta: dict[str, Any] | None = None,
) -> list[Prescription]:
    """Analyze queries for inefficient evaluation patterns.

    Args:
        queries: List of captured queries to analyze.
        models_meta: Optional model metadata (not used).

    Returns:
        List of prescriptions for detected evaluation issues.
    """
    if not queries or not self.is_enabled():
        return []

    try:
        return self._detect_eval_patterns(queries)
    except Exception:
        logger.warning("query_doctor: queryset eval analysis failed", exc_info=True)
        return []

DRFSerializerAnalyzer

Detects N+1 patterns caused by DRF serializer nesting.

DRFSerializerAnalyzer

Bases: BaseAnalyzer

Analyzer that detects DRF views missing queryset optimizations.

Inspects DRF serializer classes for nested serializers and checks whether the corresponding queryset uses select_related or prefetch_related for those relations.

analyze(queries, models_meta=None)

Analyze captured queries for DRF serializer issues.

Note: This analyzer primarily works through analyze_view() which is called with DRF-specific context. The query-based analyze() method serves as a fallback interface.

Parameters:

Name Type Description Default
queries list[CapturedQuery]

List of captured queries to analyze.

 required
models_meta dict[str, Any] | None

Optional model metadata.

 None

Returns:

Type Description
list[Prescription]

List of prescriptions (empty for query-only analysis).
Source code in src/query_doctor/analyzers/drf_serializer.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def analyze(
    self,
    queries: list[CapturedQuery],
    models_meta: dict[str, Any] | None = None,
) -> list[Prescription]:
    """Analyze captured queries for DRF serializer issues.

    Note: This analyzer primarily works through analyze_view() which
    is called with DRF-specific context. The query-based analyze()
    method serves as a fallback interface.

    Args:
        queries: List of captured queries to analyze.
        models_meta: Optional model metadata.

    Returns:
        List of prescriptions (empty for query-only analysis).
    """
    if not self.is_enabled():
        return []
    return []
analyze_view(view_class=None, serializer_class=None, queryset=None)

Analyze a DRF view for missing queryset optimizations.

Parameters:

Name Type Description Default
view_class Any

The DRF ViewSet or APIView class.

 None
serializer_class Any

The serializer class used by the view.

 None
queryset Any

The queryset used by the view.

 None

Returns:

Type Description
list[Prescription]

List of prescriptions for detected issues.
Source code in src/query_doctor/analyzers/drf_serializer.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def analyze_view(
    self,
    view_class: Any = None,
    serializer_class: Any = None,
    queryset: Any = None,
) -> list[Prescription]:
    """Analyze a DRF view for missing queryset optimizations.

    Args:
        view_class: The DRF ViewSet or APIView class.
        serializer_class: The serializer class used by the view.
        queryset: The queryset used by the view.

    Returns:
        List of prescriptions for detected issues.
    """
    if not self.is_enabled() or serializer_class is None or queryset is None:
        return []

    try:
        return self._detect_missing_optimizations(view_class, serializer_class, queryset)
    except Exception:
        logger.warning("query_doctor: DRF serializer analysis failed", exc_info=True)
        return []

QueryComplexityAnalyzer

Scores queries by complexity and flags those above threshold.

QueryComplexityAnalyzer

Bases: BaseAnalyzer

Analyzes SQL queries for excessive complexity.

Scores queries based on structural patterns (JOINs, subqueries, aggregations, etc.) and flags those exceeding a configurable threshold.

analyze(queries, models_meta=None)

Analyze captured queries for excessive complexity.

Parameters:

Name Type Description Default
queries list[CapturedQuery]

List of captured SQL queries to analyze.

 required
models_meta dict[str, Any] | None

Optional Django model metadata (unused).

 None

Returns:

Type Description
list[Prescription]

List of Prescription objects for overly complex queries.
Source code in src/query_doctor/analyzers/complexity.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def analyze(
    self,
    queries: list[CapturedQuery],
    models_meta: dict[str, Any] | None = None,
) -> list[Prescription]:
    """Analyze captured queries for excessive complexity.

    Args:
        queries: List of captured SQL queries to analyze.
        models_meta: Optional Django model metadata (unused).

    Returns:
        List of Prescription objects for overly complex queries.
    """
    if not self.is_enabled():
        return []

    config = get_config()
    analyzer_conf = config.get("ANALYZERS", {}).get("complexity", {})
    threshold = analyzer_conf.get("threshold", 8)

    prescriptions: list[Prescription] = []

    for q in queries:
        if not q.is_select:
            continue

        score = self._score_complexity(q.normalized_sql)
        if score >= threshold:
            severity = Severity.CRITICAL if score >= 12 else Severity.WARNING
            prescriptions.append(
                Prescription(
                    issue_type=IssueType.QUERY_COMPLEXITY,
                    severity=severity,
                    description=(
                        f"Query complexity score {score} exceeds threshold {threshold}"
                    ),
                    fix_suggestion=self._suggest_simplification(q.normalized_sql, score),
                    callsite=q.callsite,
                )
            )

    return prescriptions

Reporters

ConsoleReporter

Terminal output with Rich formatting and plain-text fallback.

ConsoleReporter(stream=None, group_by=None)

Formats and prints diagnosis reports to the console.

Uses Rich for styled output if available, otherwise plain text. Supports grouped output mode for related prescriptions.

Initialize the console reporter.

Parameters:

Name Type Description Default
stream Any

Output stream (file-like object). Defaults to sys.stderr. Accepts TextIO, Django's OutputWrapper, or any writable stream.

 None
group_by str | None

If set, group prescriptions by this strategy ("file_analyzer", "root_cause", "view").

 None
Source code in src/query_doctor/reporters/console.py 
28
29
30
31
32
33
34
35
36
37
38
def __init__(self, stream: Any = None, group_by: str | None = None) -> None:
    """Initialize the console reporter.

    Args:
        stream: Output stream (file-like object). Defaults to sys.stderr.
                Accepts TextIO, Django's OutputWrapper, or any writable stream.
        group_by: If set, group prescriptions by this strategy
                  ("file_analyzer", "root_cause", "view").
    """
    self._stream = stream or sys.stderr
    self._group_by = group_by
render(report)

Render a diagnosis report as a formatted string.

Parameters:

Name Type Description Default
report DiagnosisReport

The diagnosis report to render.

 required

Returns:

Type Description
str

Formatted string representation of the report.
Source code in src/query_doctor/reporters/console.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
def render(self, report: DiagnosisReport) -> str:
    """Render a diagnosis report as a formatted string.

    Args:
        report: The diagnosis report to render.

    Returns:
        Formatted string representation of the report.
    """
    try:
        return self._render_rich(report)
    except ImportError:
        return self._render_plain(report)
report(report)

Print the diagnosis report to the configured stream.

If group_by was set during init, groups related prescriptions.

Parameters:

Name Type Description Default
report DiagnosisReport

The diagnosis report to print.

 required
Source code in src/query_doctor/reporters/console.py 
54
55
56
57
58
59
60
61
62
63
def report(self, report: DiagnosisReport) -> None:
    """Print the diagnosis report to the configured stream.

    If group_by was set during init, groups related prescriptions.

    Args:
        report: The diagnosis report to print.
    """
    output = self._render_grouped(report) if self._group_by else self.render(report)
    print(output, file=self._stream)

JSONReporter

Structured JSON output for CI/CD pipelines.

JSONReporter(output_path=None)

Formats diagnosis reports as structured JSON.

Optionally writes the JSON to a file for CI/CD integration.

Initialize the JSON reporter.

Parameters:

Name Type Description Default
output_path str | None

Optional file path to write JSON output. If None, output is only available via render().

 None
Source code in src/query_doctor/reporters/json_reporter.py 
25
26
27
28
29
30
31
32
def __init__(self, output_path: str | None = None) -> None:
    """Initialize the JSON reporter.

    Args:
        output_path: Optional file path to write JSON output.
                     If None, output is only available via render().
    """
    self._output_path = output_path
render(report)

Render a diagnosis report as a JSON string.

Parameters:

Name Type Description Default
report DiagnosisReport

The diagnosis report to render.

 required

Returns:

Type Description
str

JSON string representation of the report.
Source code in src/query_doctor/reporters/json_reporter.py 
34
35
36
37
38
39
40
41
42
43
44
def render(self, report: DiagnosisReport) -> str:
    """Render a diagnosis report as a JSON string.

    Args:
        report: The diagnosis report to render.

    Returns:
        JSON string representation of the report.
    """
    data = self._build_json(report)
    return json.dumps(data, indent=2, default=str)
report(report)

Write the diagnosis report to the configured output path.

Parameters:

Name Type Description Default
report DiagnosisReport

The diagnosis report to output.

 required
Source code in src/query_doctor/reporters/json_reporter.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def report(self, report: DiagnosisReport) -> None:
    """Write the diagnosis report to the configured output path.

    Args:
        report: The diagnosis report to output.
    """
    output = self.render(report)

    if self._output_path:
        try:
            with open(self._output_path, "w", encoding="utf-8") as f:
                f.write(output)
        except OSError:
            logger.warning(
                "query_doctor: failed to write JSON report to %s",
                self._output_path,
                exc_info=True,
            )

HTMLReporter

Interactive HTML dashboard report.

HTMLReporter(output_path=None)

Generates standalone HTML reports for query diagnosis.

Produces a single HTML file with inline CSS suitable for saving, sharing, or viewing in a browser.

Initialize the HTML reporter.

Parameters:

Name Type Description Default
output_path str | None

Optional file path to write HTML output.

 None
Source code in src/query_doctor/reporters/html_reporter.py 
36
37
38
39
40
41
42
def __init__(self, output_path: str | None = None) -> None:
    """Initialize the HTML reporter.

    Args:
        output_path: Optional file path to write HTML output.
    """
    self._output_path = output_path
render(report)

Render a diagnosis report as a standalone HTML string.

Parameters:

Name Type Description Default
report DiagnosisReport

The diagnosis report to render.

 required

Returns:

Type Description
str

Complete HTML document as a string.
Source code in src/query_doctor/reporters/html_reporter.py 
44
45
46
47
48
49
50
51
52
53
def render(self, report: DiagnosisReport) -> str:
    """Render a diagnosis report as a standalone HTML string.

    Args:
        report: The diagnosis report to render.

    Returns:
        Complete HTML document as a string.
    """
    return self._build_html(report)
report(report)

Write the diagnosis report to the configured output path.

Parameters:

Name Type Description Default
report DiagnosisReport

The diagnosis report to output.

 required
Source code in src/query_doctor/reporters/html_reporter.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def report(self, report: DiagnosisReport) -> None:
    """Write the diagnosis report to the configured output path.

    Args:
        report: The diagnosis report to output.
    """
    output = self.render(report)

    if self._output_path:
        try:
            with open(self._output_path, "w", encoding="utf-8") as f:
                f.write(output)
        except OSError:
            logger.warning(
                "query_doctor: failed to write HTML report to %s",
                self._output_path,
                exc_info=True,
            )

LogReporter

Python logging integration for production monitoring.

LogReporter

Sends diagnosis reports to Python's logging system.

Each prescription is logged at the appropriate level based on severity.

report(report)

Log the diagnosis report.

Parameters:

Name Type Description Default
report DiagnosisReport

The diagnosis report to log.

 required
Source code in src/query_doctor/reporters/log_reporter.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def report(self, report: DiagnosisReport) -> None:
    """Log the diagnosis report.

    Args:
        report: The diagnosis report to log.
    """
    logger.info(
        "Query Doctor: %d queries, %.1fms, %d issues",
        report.total_queries,
        report.total_time_ms,
        report.issues,
    )

    for prescription in report.prescriptions:
        self._log_prescription(prescription)

OTelReporter

OpenTelemetry span export for observability platforms.

OTelReporter(tracer=None)

Reports query diagnosis data via OpenTelemetry spans.

Creates a span for each diagnosis run with attributes for summary metrics and events for each prescription. Sets span status to ERROR if critical issues are found.

If OpenTelemetry is not installed, all operations are no-ops.

Initialize the OTel reporter.

Parameters:

Name Type Description Default
tracer Any

Optional pre-configured OTel tracer. If None, one is created from the global TracerProvider.

 None
Source code in src/query_doctor/reporters/otel_exporter.py 
44
45
46
47
48
49
50
51
52
def __init__(self, tracer: Any = None) -> None:
    """Initialize the OTel reporter.

    Args:
        tracer: Optional pre-configured OTel tracer. If None, one is
                created from the global TracerProvider.
    """
    self._tracer = tracer
    self._has_otel = HAS_OTEL or tracer is not None
has_otel property

Whether OpenTelemetry is available.

report(report)

Export diagnosis report as OTel span data.

Parameters:

Name Type Description Default
report DiagnosisReport

The diagnosis report to export.

 required
Source code in src/query_doctor/reporters/otel_exporter.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def report(self, report: DiagnosisReport) -> None:
    """Export diagnosis report as OTel span data.

    Args:
        report: The diagnosis report to export.
    """
    if not self._has_otel:
        return

    try:
        self._export(report)
    except Exception:
        logger.warning(
            "query_doctor: OpenTelemetry export failed",
            exc_info=True,
        )

Configuration

get_config

Access django-query-doctor settings with defaults.

get_config() cached

Return the merged configuration, cached after first call.

Reads the QUERY_DOCTOR setting from Django settings and deep-merges it with DEFAULT_CONFIG. The result is cached for performance.

Source code in src/query_doctor/conf.py 
61
62
63
64
65
66
67
68
69
70
71
@functools.lru_cache(maxsize=1)
def get_config() -> dict[str, Any]:
    """Return the merged configuration, cached after first call.

    Reads the QUERY_DOCTOR setting from Django settings and deep-merges
    it with DEFAULT_CONFIG. The result is cached for performance.
    """
    from django.conf import settings

    user_config: dict[str, Any] = getattr(settings, "QUERY_DOCTOR", {})
    return _deep_merge(DEFAULT_CONFIG, user_config)

Fingerprinting

SQL Fingerprinting

Normalize and hash SQL statements for grouping.

fingerprint

SQL normalization and fingerprinting for query pattern detection.

Provides functions to normalize SQL queries (replacing literals with placeholders), generate deterministic fingerprints for grouping similar queries, and extract table names from SQL statements.

extract_tables(sql)

Extract table names from FROM and JOIN clauses in a SQL query.

Handles: FROM table, JOIN table, FROM table AS alias, FROM "quoted_table", and subqueries. Returns a deduplicated list of table names (without quotes or aliases).

Source code in src/query_doctor/fingerprint.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def extract_tables(sql: str) -> list[str]:
    """Extract table names from FROM and JOIN clauses in a SQL query.

    Handles: FROM table, JOIN table, FROM table AS alias,
    FROM "quoted_table", and subqueries.
    Returns a deduplicated list of table names (without quotes or aliases).
    """
    # Match FROM or JOIN followed by a table name (optionally quoted)
    matches = _RE_FROM_JOIN_TABLE.findall(sql)

    # Deduplicate while preserving order
    seen: set[str] = set()
    result: list[str] = []
    for table in matches:
        if table not in seen:
            seen.add(table)
            result.append(table)

    return result
fingerprint(sql)

Generate a SHA-256 fingerprint (first 16 hex chars) for a SQL query.

Two queries with the same structure but different parameter values will produce the same fingerprint.

Source code in src/query_doctor/fingerprint.py 
64
65
66
67
68
69
70
71
def fingerprint(sql: str) -> str:
    """Generate a SHA-256 fingerprint (first 16 hex chars) for a SQL query.

    Two queries with the same structure but different parameter values
    will produce the same fingerprint.
    """
    normalized = normalize_sql(sql)
    return hashlib.sha256(normalized.encode()).hexdigest()[:16]
normalize_sql(sql)

Normalize a SQL query by replacing literals with placeholders.

Replaces quoted strings, numbers, booleans, and IN-clause lists with '?', collapses whitespace, strips semicolons, and lowercases everything.

Source code in src/query_doctor/fingerprint.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def normalize_sql(sql: str) -> str:
    """Normalize a SQL query by replacing literals with placeholders.

    Replaces quoted strings, numbers, booleans, and IN-clause lists with '?',
    collapses whitespace, strips semicolons, and lowercases everything.
    """
    result = sql

    # Replace single-quoted strings (including escaped quotes inside)
    result = _RE_QUOTED_STRING.sub("?", result)

    # Replace Django-style %s parameter placeholders
    result = _RE_PARAM_PLACEHOLDER.sub("?", result)

    # Replace boolean literals (standalone TRUE/FALSE)
    result = _RE_TRUE.sub("?", result)
    result = _RE_FALSE.sub("?", result)

    # Replace numeric literals (integers and floats)
    # Must come after string replacement to avoid matching numbers inside strings
    result = _RE_NUMERIC.sub("?", result)

    # Collapse IN (...) lists to IN (?)
    result = _RE_IN_CLAUSE.sub("IN (?)", result)

    # Collapse whitespace (spaces, tabs, newlines) to single space
    result = _RE_WHITESPACE.sub(" ", result)

    # Strip trailing semicolons
    result = result.rstrip(";")

    # Strip leading/trailing whitespace
    result = result.strip()

    # Lowercase everything
    result = result.lower()

    return result

Stack Tracing

Source Code Mapping

Map captured queries back to user source code locations.

stack_tracer

Stack trace capture for mapping SQL queries to user source code.

Walks the call stack to find the first frame in user code (filtering out Django internals, this package, and stdlib modules) so each query can be attributed to a specific file:line in the application.

capture_callsite(exclude_modules=None)

Walk the stack and find the first frame in user code.

Filters out frames from query_doctor, Django internals, and stdlib. Returns the last remaining frame (closest to the query trigger), or None if no user code frame is found.

Source code in src/query_doctor/stack_tracer.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def capture_callsite(
    exclude_modules: list[str] | None = None,
) -> CallSite | None:
    """Walk the stack and find the first frame in user code.

    Filters out frames from query_doctor, Django internals, and stdlib.
    Returns the last remaining frame (closest to the query trigger),
    or None if no user code frame is found.
    """
    try:
        stack = traceback.extract_stack()
        if exclude_modules:
            exclude = _DEFAULT_EXCLUDE_PATTERNS + list(exclude_modules)
        else:
            exclude = _DEFAULT_EXCLUDE_PATTERNS

        # Filter frames to find user code
        user_frames = []
        for frame in stack:
            filename = frame.filename
            # Skip frames matching any exclude pattern
            if any(pattern in filename for pattern in exclude):
                continue
            # Skip frames from pytest/pluggy internals
            if any(p in filename for p in ["_pytest", "pluggy", "site-packages", "runpy.py"]):
                continue
            user_frames.append(frame)

        if not user_frames:
            return None

        # Take the last user-code frame (closest to the call site)
        frame = user_frames[-1]

        # Try to read the actual source line
        line_no = frame.lineno or 0
        code_context = linecache.getline(frame.filename, line_no).strip()

        return CallSite(
            filepath=frame.filename,
            line_number=line_no,
            function_name=frame.name,
            code_context=code_context,
        )
    except Exception:
        logger.warning("query_doctor: failed to capture callsite", exc_info=True)
        return None

Exceptions

All exceptions raised by django-query-doctor inherit from QueryDoctorError:

exceptions

Exception hierarchy for django-query-doctor.

All package exceptions inherit from QueryDoctorError to allow callers to catch any query-doctor-specific error with a single except clause.

AnalyzerError

Bases: QueryDoctorError

Raised when an analyzer encounters an error.

ConfigError

Bases: QueryDoctorError

Raised when there is a configuration error.

InterceptorError

Bases: QueryDoctorError

Raised when the query interceptor encounters an error.

QueryBudgetError(message, report=None)

Bases: QueryDoctorError

Raised when a function exceeds its query budget.

Attributes:

Name Type Description
report

The DiagnosisReport from the function execution.

Initialize with a message and optional report.

Parameters:

Name Type Description Default
message str

Human-readable description of the budget violation.

 required
report DiagnosisReport | None

The DiagnosisReport from the function execution.

 None
Source code in src/query_doctor/exceptions.py 
38
39
40
41
42
43
44
45
46
def __init__(self, message: str, report: DiagnosisReport | None = None) -> None:
    """Initialize with a message and optional report.

    Args:
        message: Human-readable description of the budget violation.
        report: The DiagnosisReport from the function execution.
    """
    super().__init__(message)
    self.report = report
QueryDoctorError

Bases: Exception

Base exception for all django-query-doctor errors.