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.

Works under both WSGI and ASGI. The class declares async_capable = False deliberately: Django then adapts it with sync_to_async(thread_sensitive=True), which runs it in the same thread-sensitive executor Django runs all ORM work in. Because connections["default"] is thread-local, that co-location is what lets the execute_wrapper installed here see the queries the view issues. Declaring async_capable = True places the middleware on the event loop thread instead, where it wraps a connection object the ORM never touches and captures nothing. This is ordinary behaviour for any sync-capable middleware under ASGI, not a query-doctor compromise.

Initialize the middleware.

The async predicate comes from asgiref.sync, which is the one Django itself uses. inspect.iscoroutinefunction does not recognise asgiref-wrapped callables before Python 3.12, so it would put a directly-instantiated middleware on the sync path and run the analysis stage before the view body.

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 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def __init__(self, get_response: Callable[..., Any]) -> None:
    """Initialize the middleware.

    The async predicate comes from ``asgiref.sync``, which is the one Django
    itself uses. ``inspect.iscoroutinefunction`` does not recognise
    asgiref-wrapped callables before Python 3.12, so it would put a
    directly-instantiated middleware on the sync path and run the analysis
    stage before the view body.

    Args:
        get_response: The next middleware or view in the chain.
    """
    self.get_response = get_response
    self._is_async = iscoroutinefunction(get_response)
    # Set once the hand-embed capture gap has been reported for this
    # instance; the wiring it describes cannot change between requests.
    self._warned_unreachable_executor = False
__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 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
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 = build_interceptor()

    from django.db import connection

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

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

    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 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
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.

Synchronous only. Entering the block from a coroutine emits a QueryDoctorWarning and captures nothing -- use the middleware for async views. See docs/guides/async-support.md.

Usage

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

Source code in src/query_doctor/context_managers.py 
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
82
83
84
85
86
87
88
89
90
91
92
93
94
@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.

    Synchronous only. Entering the block from a coroutine emits a
    ``QueryDoctorWarning`` and captures nothing -- use the middleware for
    async views. See ``docs/guides/async-support.md``.

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

    report = DiagnosisReport()
    interceptor = build_interceptor()

    from django.db import connection

    try:
        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 and apply .queryignore filtering via the shared pipeline.
            report.prescriptions = pipeline_analyze(queries, source="context_manager")
        except Exception:
            logger.warning("query_doctor: context manager analysis failed", exc_info=True)
    finally:
        # The report keeps the queries; the context must not keep a second
        # reference to them once the block is done with.
        interceptor.release()

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

Reserved, and currently always None. Nothing in the package passes it: the single call site, pipeline.analyze(), calls analyzer.analyze(queries) for every analyzer on every run. Do not write a plugin that depends on receiving a value here. Accept the parameter so your signature stays compatible; removing it from the contract would be a breaking change and is deferred to 3.0.

 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
36
37
38
39
40
41
@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: Reserved, and **currently always None**. Nothing in
            the package passes it: the single call site,
            ``pipeline.analyze()``, calls ``analyzer.analyze(queries)``
            for every analyzer on every run. Do not write a plugin that
            depends on receiving a value here. Accept the parameter so
            your signature stays compatible; removing it from the
            contract would be a breaking change and is deferred to 3.0.

    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 
43
44
45
46
47
48
49
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 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
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 or not self.is_enabled():
        return []

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

DuplicateAnalyzer

Detects exact-duplicate queries (same SQL and same bound parameters) executed more than once within a single request.

DuplicateAnalyzer

Bases: BaseAnalyzer

Analyzer that detects exact-duplicate queries.

Groups queries by a hash of SQL text + params. A query executed with the same SQL and the same parameter values more than threshold times is flagged. Queries with the same SQL structure but different parameter values are not considered duplicates by this analyzer.

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 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
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 or not self.is_enabled():
        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 or not self.is_enabled():
        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 
63
64
65
66
67
68
69
70
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 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
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 []

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
82
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),
                    callsite=q.callsite,
                    fingerprint=q.fingerprint,
                )
            )

    return prescriptions

WriteNPlusOneAnalyzer

Detects repeated single-row writes and prescribes the bulk equivalent.

WriteNPlusOneAnalyzer

Bases: BaseAnalyzer

Analyzer that detects repeated single-row writes.

Groups non-SELECT queries by fingerprint and reports groups at or over the configured threshold, prescribing the bulk statement that replaces them.

analyze(queries, models_meta=None)

Analyze queries for repeated single-row writes.

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 write N+1 issues.
Source code in src/query_doctor/analyzers/write_nplusone.py 
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
def analyze(
    self,
    queries: list[CapturedQuery],
    models_meta: dict[str, Any] | None = None,
) -> list[Prescription]:
    """Analyze queries for repeated single-row writes.

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

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

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

SerializerMethodAnalyzer

Static AST analysis of DRF SerializerMethodField bodies. Unlike the seven analyzers above it does not read captured queries: analyze() conforms to the BaseAnalyzer interface and always returns []. Drive it through analyze_serializer(), or through the check_serializers management command.

SerializerMethodAnalyzer

Bases: BaseAnalyzer

Analyzes DRF serializer classes for SerializerMethodField methods.

Detects methods that may cause N+1 queries. This is a STATIC analyzer -- it reads source code, not runtime queries. It should be invoked separately from the runtime query interception pipeline via the analyze_serializer() method.

Inherits BaseAnalyzer for plugin API compatibility. The analyze() method (required by BaseAnalyzer) returns an empty list since this analyzer operates on serializer classes, not captured queries. Use analyze_serializer() for actual analysis.

analyze(queries, models_meta=None)

Conform to BaseAnalyzer interface. Returns empty list.

This is a static analyzer that operates on serializer classes, not runtime queries. Use analyze_serializer() instead.

Parameters:

Name Type Description Default
queries list[CapturedQuery]

Ignored -- this analyzer does not use captured queries.

 required
models_meta dict[str, Any] | None

Ignored.

 None

Returns:

Type Description
list[Prescription]

Always an empty list.
Source code in src/query_doctor/analyzers/serializer_method.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def analyze(
    self,
    queries: list[CapturedQuery],
    models_meta: dict[str, Any] | None = None,
) -> list[Prescription]:
    """Conform to BaseAnalyzer interface. Returns empty list.

    This is a static analyzer that operates on serializer classes, not
    runtime queries. Use ``analyze_serializer()`` instead.

    Args:
        queries: Ignored -- this analyzer does not use captured queries.
        models_meta: Ignored.

    Returns:
        Always an empty list.
    """
    return []
analyze_serializer(serializer_cls)

Analyze a single serializer class for N+1 patterns.

Finds all SerializerMethodField declarations, locates the corresponding get_ methods, parses them with ast, and walks the AST looking for dangerous patterns.

Parameters:

Name Type Description Default
serializer_cls Any

A DRF serializer class to analyze.

 required

Returns:

Type Description
list[Prescription]

List of Prescription objects describing detected issues.
Source code in src/query_doctor/analyzers/serializer_method.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def analyze_serializer(self, serializer_cls: Any) -> list[Prescription]:
    """Analyze a single serializer class for N+1 patterns.

    Finds all SerializerMethodField declarations, locates the corresponding
    get_<field> methods, parses them with ast, and walks the AST looking
    for dangerous patterns.

    Args:
        serializer_cls: A DRF serializer class to analyze.

    Returns:
        List of Prescription objects describing detected issues.
    """
    if not self.is_enabled():
        return []

    try:
        from rest_framework import serializers as drf_serializers
    except ImportError:
        logger.debug("DRF not installed, skipping SerializerMethodAnalyzer")
        return []

    prescriptions: list[Prescription] = []
    method_fields = self._find_method_fields(serializer_cls, drf_serializers)

    for field_name in method_fields:
        method_name = f"get_{field_name}"
        method = self._find_method(serializer_cls, method_name)
        if method is None:
            continue

        try:
            source = inspect.getsource(method)
            source = textwrap.dedent(source)
            tree = ast.parse(source)
        except (OSError, TypeError, SyntaxError):
            logger.debug(
                "Could not parse source for %s.%s",
                serializer_cls.__name__,
                method_name,
            )
            continue

        # Determine the parameter name for the serialized object
        obj_param = self._get_obj_param(tree)
        if obj_param is None:
            continue

        try:
            source_file = inspect.getfile(method)
            source_lines = inspect.getsourcelines(method)
            start_line = source_lines[1]
        except (OSError, TypeError):
            source_file = "<unknown>"
            start_line = 0

        issues = self._walk_method(tree, obj_param, field_name, serializer_cls)
        for issue in issues:
            callsite = CallSite(
                filepath=source_file,
                line_number=start_line + issue.get("line_offset", 0),
                function_name=method_name,
                code_context=issue.get("code_context", ""),
            )
            prescriptions.append(
                Prescription(
                    issue_type=IssueType.SERIALIZER_METHOD_FIELD,
                    severity=issue.get("severity", Severity.WARNING),
                    description=issue["description"],
                    fix_suggestion=issue["fix_suggestion"],
                    callsite=callsite,
                    extra={
                        "field": field_name,
                        "pattern": issue["pattern"],
                        "serializer": serializer_cls.__name__,
                    },
                )
            )

    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 
47
48
49
50
51
52
53
54
55
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 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
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 the table names a SQL statement reads or writes.

Handles: FROM table, JOIN table, FROM table AS alias, FROM "quoted_table", and subqueries; plus the write forms UPDATE table, INSERT INTO table and DELETE FROM table. Returns a deduplicated list of table names (without quotes or aliases).

Source code in src/query_doctor/fingerprint.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def extract_tables(sql: str) -> list[str]:
    """Extract the table names a SQL statement reads or writes.

    Handles: FROM table, JOIN table, FROM table AS alias,
    FROM "quoted_table", and subqueries; plus the write forms
    UPDATE table, INSERT INTO table and DELETE FROM table.
    Returns a deduplicated list of table names (without quotes or aliases).
    """
    # Match the keywords that introduce 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 
70
71
72
73
74
75
76
77
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 
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
63
64
65
66
67
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 
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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
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 installed packages and test-runner internals
            if any(p in filename for p in _INSTALL_PATH_PATTERNS):
                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 and warning category 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. Runtime advisories are emitted as QueryDoctorWarning so they can be filtered by category without touching other UserWarnings.

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.

QueryDoctorWarning

Bases: UserWarning

Base warning category for query_doctor runtime advisories.

Emitted where query_doctor detects usage that silently does not do what it appears to do. Suppress this category alone with -W ignore::query_doctor.QueryDoctorWarning (or the equivalent filterwarnings entry) without silencing other UserWarnings.