Coverage for src/asgi_profiler/models.py: 99%
234 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-22 16:59 +0000
1"""Recorded data structures."""
3from __future__ import annotations
5import math
6from collections import Counter
7from dataclasses import dataclass, field
8from datetime import datetime, timezone
9from typing import Any
11from .fingerprint import is_truncated, sql_hash
13#: Statements longer than this are stored truncated. An ORM bulk insert can
14#: produce a single statement tens of kilobytes long, and `max_requests`
15#: bounds the number of requests kept, not their size -- without a cap one
16#: pathological endpoint can hold hundreds of megabytes in the ring buffer.
17MAX_SQL_CHARS = 4000
20@dataclass(slots=True)
21class Query:
22 """One SQL statement executed during a request."""
24 sql: str
25 params: str
26 duration_ms: float
27 stack: list[str] = field(default_factory=list)
28 #: Milliseconds from the start of the request to the start of this
29 #: statement. Two statements overlap when one starts before the other
30 #: ends, which is the whole basis of telling "six queries that ran one
31 #: after another" apart from "six queries that ran at once" -- the first
32 #: is worth 300 ms of someone's afternoon and the second is not.
33 started_ms: float = 0.0
34 #: The statement ran on the event loop thread, outside SQLAlchemy's async
35 #: bridge: a synchronous driver call that stopped the loop dead for its
36 #: whole duration. See :mod:`asgi_profiler.detectors.blocking`.
37 blocking: bool = False
38 #: Set by :meth:`Profile.finalise` once the whole request is known.
39 is_duplicate: bool = False
40 #: The exception text, when the statement failed. `None` means it succeeded.
41 error: str | None = None
43 @property
44 def failed(self) -> bool:
45 return self.error is not None
47 @property
48 def ended_ms(self) -> float:
49 return self.started_ms + self.duration_ms
51 @property
52 def sql_hash(self) -> str:
53 """Identity of the statement, ignoring per-execution values."""
54 return sql_hash(self.sql)
56 @property
57 def truncated(self) -> bool:
58 return is_truncated(self.sql)
60 @property
61 def operation(self) -> str:
62 head = self.sql.lstrip().split(" ", 1)[0].upper()
63 return head if head.isalpha() else "SQL"
66@dataclass(slots=True)
67class Problem:
68 """Something the detectors concluded about a request.
70 The difference between a profiler and a report: `duplicate_count = 11` is
71 a number the reader has to interpret, and "11 identical queries issued
72 from repository.py:20, 57.6 ms -- load the relationship instead" is a
73 finding they can act on.
75 `fingerprint` is the identity of the *problem*, not of the request: the
76 same N+1 seen on four hundred requests has one fingerprint, which is what
77 lets the viewer collapse them into a single row and what lets CI say "this
78 is not new".
79 """
81 type: str
82 fingerprint: str
83 title: str
84 detail: str = ""
85 #: Named facts for the template: the source statement, the offending
86 #: frame, how many times it repeated. Display data, not API.
87 evidence: dict[str, Any] = field(default_factory=dict)
88 #: Indices into :attr:`Profile.queries`, so the detail page can highlight
89 #: the statements this is about without storing them twice.
90 offenders: list[int] = field(default_factory=list)
91 #: Time attributable to the problem. Not always time you would get back by
92 #: fixing it -- for consecutive queries that is a separate number -- but
93 #: always the size of the thing being pointed at.
94 time_ms: float = 0.0
96 def as_dict(self) -> dict[str, Any]:
97 return {
98 "type": self.type,
99 "fingerprint": self.fingerprint,
100 "title": self.title,
101 "detail": self.detail,
102 "evidence": self.evidence,
103 "offenders": self.offenders,
104 "time_ms": round(self.time_ms, 3),
105 }
107 @classmethod
108 def from_dict(cls, data: dict[str, Any]) -> Problem:
109 return cls(
110 type=str(data.get("type", "")),
111 fingerprint=str(data.get("fingerprint", "")),
112 title=str(data.get("title", "")),
113 detail=str(data.get("detail", "")),
114 evidence=dict(data.get("evidence") or {}),
115 offenders=list(data.get("offenders") or []),
116 time_ms=float(data.get("time_ms") or 0.0),
117 )
120@dataclass(slots=True)
121class QueryGroup:
122 """Identical statements within one request, collapsed into a single row.
124 A 500-row N+1 renders 500 near-identical list items with the same stack
125 repeated under each -- 447 kB of HTML that you have to scroll past to
126 find the line that caused it. Collapsing is the difference between
127 "I saw the N+1" and "I found it".
129 Grouped by :attr:`Query.sql_hash` rather than by literal text, so a page
130 of ten rows and a page of eleven -- identical queries with different `IN`
131 lists -- are one row here instead of two.
132 """
134 sql: str
135 sql_hash: str = ""
136 count: int = 0
137 total_ms: float = 0.0
138 max_ms: float = 0.0
139 #: The first occurrence, for its stack and parameters.
140 first: Query | None = None
141 #: The first occurrence that *failed*, if any. Kept separately because the
142 #: failure is rarely the first execution, and showing the first execution's
143 #: (empty) error next to a "failed" badge hides the only thing you needed.
144 first_failure: Query | None = None
145 #: A sample of the distinct parameter renderings, for display.
146 params: list[str] = field(default_factory=list)
147 #: How many distinct renderings there actually were, so the "and N more"
148 #: hint counts what was elided rather than what was executed.
149 distinct_params: int = 0
150 failures: int = 0
151 #: How many of the executions blocked the event loop.
152 blocking: int = 0
153 #: Ordinal positions in the request, so the order of execution is legible.
154 positions: list[int] = field(default_factory=list)
155 #: Distinct stacks. More than one means the same SQL was issued from
156 #: different places, and one shared stack would be a lie.
157 call_sites: int = 0
159 @property
160 def is_duplicate(self) -> bool:
161 return self.count > 1
163 @property
164 def operation(self) -> str:
165 return self.first.operation if self.first else "SQL"
167 @property
168 def _representative(self) -> Query | None:
169 return self.first_failure or self.first
171 @property
172 def stack(self) -> list[str]:
173 query = self._representative
174 return query.stack if query else []
176 @property
177 def error(self) -> str | None:
178 return self.first_failure.error if self.first_failure else None
180 @property
181 def failed(self) -> bool:
182 return self.failures > 0
184 @property
185 def avg_ms(self) -> float:
186 return self.total_ms / self.count if self.count else 0.0
189#: Parameter renderings that carry no information and are not worth the row.
190_EMPTY_PARAMS = ("()", "{}", "[]", "", "None")
192#: How many distinct parameter renderings and positions to keep per group.
193_PARAM_SAMPLE = 5
194_POSITION_SAMPLE = 50
197def group_queries(queries: list[Query]) -> list[QueryGroup]:
198 """Collapse repeated statements, preserving first-execution order."""
199 groups: dict[str, QueryGroup] = {}
200 seen_params: dict[str, set[str]] = {}
201 seen_stacks: dict[str, set[tuple[str, ...]]] = {}
202 for position, query in enumerate(queries, start=1):
203 key = query.sql_hash
204 group = groups.get(key)
205 if group is None:
206 group = groups[key] = QueryGroup(sql=query.sql, sql_hash=key, first=query)
207 seen_params[key] = set()
208 seen_stacks[key] = set()
209 group.count += 1
210 group.total_ms += query.duration_ms
211 group.max_ms = max(group.max_ms, query.duration_ms)
212 if query.blocking:
213 group.blocking += 1
214 if query.failed:
215 group.failures += 1
216 if group.first_failure is None:
217 group.first_failure = query
218 if len(group.positions) < _POSITION_SAMPLE:
219 group.positions.append(position)
220 if query.params not in _EMPTY_PARAMS:
221 distinct = seen_params[key]
222 if query.params not in distinct:
223 distinct.add(query.params)
224 group.distinct_params = len(distinct)
225 if len(group.params) < _PARAM_SAMPLE:
226 group.params.append(query.params)
227 if query.stack:
228 stacks = seen_stacks[key]
229 stacks.add(tuple(query.stack))
230 group.call_sites = len(stacks)
231 return list(groups.values())
234@dataclass(slots=True)
235class Profile:
236 """One HTTP request.
238 The counters at the bottom are filled in by :meth:`finalise`, which the
239 middleware calls once when the request ends. They are stored rather than
240 computed on access because the viewer reads them once per row per render,
241 and because a storage backend needs to filter and group on them without
242 rehydrating every statement.
243 """
245 id: str
246 method: str
247 path: str
248 #: The matched route *pattern* -- ``/users/{user_id}`` rather than
249 #: ``/users/42``. Empty when routing did not expose one.
250 route: str = ""
251 query_string: str = ""
252 status_code: int = 0
253 duration_ms: float = 0.0
254 recorded_at: datetime = field(default_factory=lambda: datetime.now(tz=timezone.utc))
255 request_headers: dict[str, str] = field(default_factory=dict)
256 response_headers: dict[str, str] = field(default_factory=dict)
257 client: str = ""
258 queries: list[Query] = field(default_factory=list)
259 #: What the detectors concluded. Filled by the middleware after
260 #: :meth:`finalise`, because a detector needs configured thresholds and a
261 #: dataclass should not reach for configuration.
262 problems: list[Problem] = field(default_factory=list)
264 # -- derived, filled by finalise() ------------------------------------
265 query_count: int = 0
266 query_ms: float = 0.0
267 #: How many queries repeat a statement already seen in this request. The
268 #: N+1 signal: a relationship loaded once per row produces the same
269 #: statement many times over.
270 duplicate_count: int = 0
271 error_count: int = 0
272 #: How many statements stopped the event loop.
273 blocking_count: int = 0
275 def finalise(self) -> None:
276 """Mark duplicated statements and compute the counters.
278 Called once, when the request ends. Mutating `queries` afterwards
279 leaves the counters stale.
280 """
281 counts = Counter(q.sql_hash for q in self.queries)
282 for query in self.queries:
283 query.is_duplicate = counts[query.sql_hash] > 1
284 self.query_count = len(self.queries)
285 self.query_ms = sum(q.duration_ms for q in self.queries)
286 self.duplicate_count = sum(n - 1 for n in counts.values() if n > 1)
287 self.error_count = sum(1 for q in self.queries if q.failed)
288 self.blocking_count = sum(1 for q in self.queries if q.blocking)
290 # -- presentation -----------------------------------------------------
291 @property
292 def group(self) -> str:
293 """The label this request aggregates under: its route, else its path."""
294 return self.route or self.path
296 @property
297 def query_groups(self) -> list[QueryGroup]:
298 return group_queries(self.queries)
300 @property
301 def python_ms(self) -> float:
302 return max(0.0, self.duration_ms - self.query_ms)
304 @property
305 def full_path(self) -> str:
306 return f"{self.path}?{self.query_string}" if self.query_string else self.path
308 @property
309 def status_class(self) -> str:
310 if self.status_code >= 500:
311 return "err"
312 if self.status_code >= 400:
313 return "warn"
314 if self.status_code >= 300:
315 return "info"
316 return "ok"
318 def as_dict(self, *, with_queries: bool = True) -> dict[str, object]:
319 """A JSON-serialisable view, for the viewer's JSON endpoints.
321 This is what makes the profiler scriptable: assert in CI that an
322 endpoint stays under N queries, diff two runs, attach a trace to a
323 bug report, fail a build when a detector finds something new.
324 """
325 data: dict[str, object] = {
326 "id": self.id,
327 "method": self.method,
328 "path": self.path,
329 "route": self.route,
330 "group": self.group,
331 "query_string": self.query_string,
332 "status_code": self.status_code,
333 "duration_ms": round(self.duration_ms, 3),
334 "query_ms": round(self.query_ms, 3),
335 "python_ms": round(self.python_ms, 3),
336 "query_count": self.query_count,
337 "duplicate_count": self.duplicate_count,
338 "error_count": self.error_count,
339 "blocking_count": self.blocking_count,
340 "recorded_at": self.recorded_at.isoformat(),
341 "client": self.client,
342 "problems": [p.as_dict() for p in self.problems],
343 }
344 if with_queries:
345 data["request_headers"] = self.request_headers
346 data["response_headers"] = self.response_headers
347 data["queries"] = [
348 {
349 "sql": q.sql,
350 "sql_hash": q.sql_hash,
351 "params": q.params,
352 "duration_ms": round(q.duration_ms, 3),
353 "started_ms": round(q.started_ms, 3),
354 "is_duplicate": q.is_duplicate,
355 "blocking": q.blocking,
356 "error": q.error,
357 "stack": q.stack,
358 }
359 for q in self.queries
360 ]
361 return data
364def _percentile(ordered: list[float], fraction: float) -> float:
365 """Nearest-rank percentile. No numpy, and exact on small samples."""
366 if not ordered: 366 ↛ 367line 366 didn't jump to line 367 because the condition on line 366 was never true
367 return 0.0
368 rank = max(1, math.ceil(fraction * len(ordered)))
369 return ordered[min(rank, len(ordered)) - 1]
372@dataclass(slots=True)
373class PathSummary:
374 """Aggregate row on the summary page, keyed by route pattern where known."""
376 path: str
377 method: str
378 count: int = 0
379 total_ms: float = 0.0
380 max_ms: float = 0.0
381 total_queries: int = 0
382 total_duplicates: int = 0
383 total_errors: int = 0
384 total_problems: int = 0
385 #: Percentiles matter more than the average, which one outlier ruins and
386 #: which tells you nothing about what most users experienced.
387 p50_ms: float = 0.0
388 p95_ms: float = 0.0
389 p99_ms: float = 0.0
391 def set_percentiles(self, durations: list[float]) -> None:
392 ordered = sorted(durations)
393 self.p50_ms = _percentile(ordered, 0.50)
394 self.p95_ms = _percentile(ordered, 0.95)
395 self.p99_ms = _percentile(ordered, 0.99)
397 @property
398 def avg_ms(self) -> float:
399 return self.total_ms / self.count if self.count else 0.0
401 @property
402 def avg_queries(self) -> float:
403 return self.total_queries / self.count if self.count else 0.0
405 def as_dict(self) -> dict[str, object]:
406 return {
407 "route": self.path,
408 "method": self.method,
409 "count": self.count,
410 "total_ms": round(self.total_ms, 3),
411 "avg_ms": round(self.avg_ms, 3),
412 "p50_ms": round(self.p50_ms, 3),
413 "p95_ms": round(self.p95_ms, 3),
414 "p99_ms": round(self.p99_ms, 3),
415 "max_ms": round(self.max_ms, 3),
416 "total_queries": self.total_queries,
417 "avg_queries": round(self.avg_queries, 3),
418 "total_duplicates": self.total_duplicates,
419 "total_errors": self.total_errors,
420 "total_problems": self.total_problems,
421 }
424@dataclass(slots=True)
425class StatementSummary:
426 """One SQL statement aggregated across every request that ran it."""
428 sql: str
429 sql_hash: str = ""
430 count: int = 0
431 total_ms: float = 0.0
432 max_ms: float = 0.0
433 #: How many distinct requests ran it.
434 requests: int = 0
435 #: How many distinct routes ran it. More than one means a shared query --
436 #: often a serialiser or a permission check nobody thinks about.
437 route_count: int = 0
438 sample_route: str = ""
439 sample_method: str = "GET"
440 failures: int = 0
442 @property
443 def avg_ms(self) -> float:
444 return self.total_ms / self.count if self.count else 0.0
446 @property
447 def per_request(self) -> float:
448 return self.count / self.requests if self.requests else 0.0
450 @property
451 def operation(self) -> str:
452 head = self.sql.lstrip().split(" ", 1)[0].upper()
453 return head if head.isalpha() else "SQL"
455 def as_dict(self) -> dict[str, object]:
456 return {
457 "sql": self.sql,
458 "sql_hash": self.sql_hash,
459 "count": self.count,
460 "total_ms": round(self.total_ms, 3),
461 "avg_ms": round(self.avg_ms, 3),
462 "max_ms": round(self.max_ms, 3),
463 "requests": self.requests,
464 "per_request": round(self.per_request, 2),
465 "route_count": self.route_count,
466 "sample_route": self.sample_route,
467 "sample_method": self.sample_method,
468 "failures": self.failures,
469 }