Coverage for src/asgi_profiler/storage.py: 97%
380 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-22 16:59 +0000
1"""Where recorded profiles live.
3Two backends ship: :class:`MemoryStorage`, the zero-config default, and
4:class:`SQLiteStorage`, which survives a restart and is shared by every worker
5in a multi-process deployment.
7`Storage` is a Protocol, so a Redis or Postgres backend can be dropped in
8without touching the middleware or the viewer. Subclass :class:`BaseStorage`
9and you only have to implement the five core methods -- searching, paging,
10summarising and statement aggregation come for free, and can be overridden
11where the backend can do them better than Python can.
12"""
14from __future__ import annotations
16import atexit
17import contextlib
18import json
19import logging
20import math
21import queue
22import sqlite3
23import threading
24import time
25import urllib.parse
26import weakref
27from collections import defaultdict, deque
28from collections.abc import Iterable, Mapping
29from dataclasses import dataclass, field
30from datetime import datetime, timezone
31from pathlib import Path
32from typing import Any, Protocol
34from .models import PathSummary, Problem, Profile, Query, StatementSummary
36logger = logging.getLogger("asgi_profiler")
38#: Aliases resolved at module scope. Inside the storage classes the name
39#: `list` is bound to the method, so a bare `-> list[Profile]` annotation
40#: resolves to that method instead of the builtin -- which type checkers
41#: rightly reject, and which made the shipped `py.typed` marker a lie.
42Profiles = list[Profile]
43PathSummaries = list[PathSummary]
44StatementSummaries = list[StatementSummary]
46_ORDERINGS: dict[str, Any] = {
47 "recent": None,
48 "slowest": lambda p: -p.duration_ms,
49 "queries": lambda p: -p.query_count,
50 "sql": lambda p: -p.query_ms,
51}
54@dataclass(frozen=True)
55class Filters:
56 """The viewer's filter state, in one object both backends understand."""
58 q: str = ""
59 method: str = ""
60 status: str = "" # "ok" | "warn" | "err"
61 route: str = "" # exact match on the route pattern
62 min_ms: float | None = None
63 only_duplicates: bool = False
64 only_errors: bool = False
65 only_problems: bool = False
66 order: str = "recent"
68 @classmethod
69 def from_params(cls, params: Mapping[str, str]) -> Filters:
70 raw = (params.get("min_ms") or "").strip()
71 try:
72 min_ms = float(raw) if raw else None
73 except ValueError:
74 min_ms = None
75 if min_ms is not None and not math.isfinite(min_ms):
76 # `?min_ms=nan` compares False everywhere in Python and True
77 # nowhere in SQL, so the two backends would disagree.
78 min_ms = None
79 only = params.get("only", "")
80 order = params.get("order", "recent")
81 return cls(
82 # Capped: SQLite raises "LIKE or GLOB pattern too complex" past a
83 # few thousand characters, which would 500 the page rather than
84 # return nothing.
85 q=(params.get("q") or "").strip()[:200],
86 method=(params.get("method") or "").strip().upper(),
87 status=(params.get("status") or "").strip(),
88 route=(params.get("route") or "").strip(),
89 min_ms=min_ms,
90 only_duplicates=only == "duplicates",
91 only_errors=only == "errors",
92 only_problems=only == "problems",
93 order=order if order in _ORDERINGS else "recent",
94 )
96 @property
97 def active(self) -> bool:
98 return bool(
99 self.q
100 or self.method
101 or self.status
102 or self.route
103 or self.min_ms is not None
104 or self.only_duplicates
105 or self.only_errors
106 or self.only_problems
107 )
109 def matches(self, profile: Profile) -> bool:
110 if self.q and self.q.lower() not in profile.full_path.lower():
111 return False
112 if self.method and profile.method != self.method:
113 return False
114 if self.route and profile.group != self.route:
115 return False
116 if self.status == "err" and profile.status_code < 500:
117 return False
118 if self.status == "warn" and not (400 <= profile.status_code < 500):
119 return False
120 if self.status == "ok" and not (200 <= profile.status_code < 300):
121 return False
122 if self.only_problems and not profile.problems: 122 ↛ 123line 122 didn't jump to line 123 because the condition on line 122 was never true
123 return False
124 if self.only_duplicates and not profile.duplicate_count:
125 return False
126 if self.only_errors and not profile.error_count:
127 return False
128 return not (self.min_ms is not None and profile.duration_ms < self.min_ms)
131@dataclass
132class Page:
133 """One page of results, plus enough context to render the pager.
135 The items are *listing* rows: counters are always populated, but a backend
136 may leave `queries` and the header dicts empty because the request list
137 never shows them. Use `Storage.get()` for a complete profile.
138 """
140 items: Profiles = field(default_factory=list)
141 #: Matching the filters, across all pages.
142 total: int = 0
143 number: int = 1
144 size: int = 50
146 @property
147 def pages(self) -> int:
148 return max(1, -(-self.total // self.size)) if self.size else 1
150 @property
151 def has_previous(self) -> bool:
152 return self.number > 1
154 @property
155 def has_next(self) -> bool:
156 return self.number < self.pages
158 @property
159 def first_index(self) -> int:
160 return (self.number - 1) * self.size + 1 if self.total else 0
162 @property
163 def last_index(self) -> int:
164 return min(self.number * self.size, self.total)
167class Storage(Protocol):
168 def add(self, profile: Profile) -> None: ...
170 def get(self, profile_id: str) -> Profile | None: ...
172 def list(self, *, limit: int | None = None, offset: int = 0) -> Profiles: ...
174 def count(self) -> int: ...
176 def clear(self) -> None: ...
178 def search(self, filters: Filters, *, page: int = 1, size: int = 50) -> Page: ...
180 def summarise(self) -> PathSummaries: ...
182 def statements(self, limit: int = 100) -> StatementSummaries: ...
185class BaseStorage:
186 """Generic `search`, `summarise` and `statements` on top of `list()`.
188 Correct for any backend; a backend that can push the work down to a query
189 engine should override them.
190 """
192 def list( # pragma: no cover - overridden
193 self, *, limit: int | None = None, offset: int = 0
194 ) -> Profiles:
195 raise NotImplementedError
197 def count(self) -> int: # pragma: no cover - overridden
198 raise NotImplementedError
200 def search(self, filters: Filters, *, page: int = 1, size: int = 50) -> Page:
201 matched = [p for p in self.list() if filters.matches(p)]
202 key = _ORDERINGS.get(filters.order)
203 if key is not None:
204 matched.sort(key=key)
205 return _paginate(matched, page=page, size=size)
207 def summarise(self) -> PathSummaries:
208 return summarise(self.list())
210 def statements(self, limit: int = 100) -> StatementSummaries:
211 return aggregate_statements(self.list(), limit=limit)
214def _paginate(profiles: Profiles, *, page: int, size: int) -> Page:
215 size = max(1, size)
216 total = len(profiles)
217 pages = max(1, -(-total // size))
218 number = min(max(1, page), pages)
219 start = (number - 1) * size
220 return Page(items=profiles[start : start + size], total=total, number=number, size=size)
223class MemoryStorage(BaseStorage):
224 """Bounded ring buffer, newest first. Thread-safe.
226 Per-process: with multiple workers each holds its own history. Use
227 :class:`SQLiteStorage` if that matters.
228 """
230 def __init__(self, max_requests: int = 500) -> None:
231 self._items: deque[Profile] = deque(maxlen=max_requests)
232 self._lock = threading.Lock()
234 def add(self, profile: Profile) -> None:
235 with self._lock:
236 self._items.appendleft(profile)
238 def get(self, profile_id: str) -> Profile | None:
239 with self._lock:
240 for item in self._items:
241 if item.id == profile_id:
242 return item
243 return None
245 def list(self, *, limit: int | None = None, offset: int = 0) -> Profiles:
246 with self._lock:
247 items = list(self._items)
248 if offset:
249 items = items[offset:]
250 return items[:limit] if limit is not None else items
252 def count(self) -> int:
253 with self._lock:
254 return len(self._items)
256 def clear(self) -> None:
257 with self._lock:
258 self._items.clear()
260 def __len__(self) -> int:
261 return self.count()
264#: Stores with a live writer thread, so the interpreter can flush them on the
265#: way out. A daemon thread is killed at exit with whatever is still queued,
266#: and nothing calls `close()` for a store handed to `install()` -- without
267#: this the tail of every capture is lost, which for "profile in staging, read
268#: the file later" is exactly the part you were waiting for.
269#:
270#: One handler for the process over a weak set, rather than one registration
271#: per store: a per-store hook has to be unregistered on close, and a leaked
272#: `atexit` entry per store is its own slow problem.
273_OPEN_STORES: weakref.WeakSet[Any] = weakref.WeakSet()
276def _flush_open_stores() -> None:
277 for store in list(_OPEN_STORES):
278 with contextlib.suppress(Exception):
279 store.close()
282atexit.register(_flush_open_stores)
285class IncompatibleCapture(RuntimeError):
286 """A capture file written by an incompatible schema version."""
289#: Bumped whenever the table layout changes. A mismatch rebuilds the file
290#: rather than failing with an opaque OperationalError from inside `add()`,
291#: after the response has already gone out.
292SCHEMA_VERSION = 2
294_SCHEMA = """
295CREATE TABLE IF NOT EXISTS profiles (
296 seq INTEGER PRIMARY KEY AUTOINCREMENT,
297 id TEXT NOT NULL UNIQUE,
298 method TEXT NOT NULL,
299 path TEXT NOT NULL,
300 route TEXT NOT NULL DEFAULT '',
301 grp TEXT NOT NULL DEFAULT '',
302 query_string TEXT NOT NULL DEFAULT '',
303 search_path TEXT NOT NULL DEFAULT '',
304 status_code INTEGER NOT NULL DEFAULT 0,
305 duration_ms REAL NOT NULL DEFAULT 0,
306 query_ms REAL NOT NULL DEFAULT 0,
307 query_count INTEGER NOT NULL DEFAULT 0,
308 duplicate_count INTEGER NOT NULL DEFAULT 0,
309 error_count INTEGER NOT NULL DEFAULT 0,
310 blocking_count INTEGER NOT NULL DEFAULT 0,
311 problem_count INTEGER NOT NULL DEFAULT 0,
312 recorded_at TEXT NOT NULL,
313 client TEXT NOT NULL DEFAULT '',
314 -- Its own column rather than part of `payload`, because the request list
315 -- reads it on every row and deliberately does not load the payload: the
316 -- statements are most of a profile's bytes and the listing shows none of
317 -- them. Problems are a handful of small objects, so they can be afforded
318 -- on a page of fifty rows and the badges come for free.
319 problems TEXT NOT NULL DEFAULT '[]',
320 payload TEXT NOT NULL
321);
322CREATE INDEX IF NOT EXISTS profiles_seq ON profiles (seq DESC);
323CREATE INDEX IF NOT EXISTS profiles_group ON profiles (method, grp);
325CREATE TABLE IF NOT EXISTS statements (
326 profile_seq INTEGER NOT NULL,
327 sql TEXT NOT NULL,
328 sql_hash TEXT NOT NULL DEFAULT '',
329 grp TEXT NOT NULL DEFAULT '',
330 method TEXT NOT NULL DEFAULT 'GET',
331 duration_ms REAL NOT NULL DEFAULT 0,
332 failed INTEGER NOT NULL DEFAULT 0
333);
334CREATE INDEX IF NOT EXISTS statements_profile ON statements (profile_seq);
335-- On the hash, not the text. The text is what gets displayed; the hash is
336-- what gets grouped, and grouping a 4000-character statement by its text is
337-- an index entry the size of the statement.
338CREATE INDEX IF NOT EXISTS statements_hash ON statements (sql_hash);
340CREATE TABLE IF NOT EXISTS meta (
341 key TEXT PRIMARY KEY,
342 value INTEGER NOT NULL
343);
344INSERT OR IGNORE INTO meta (key, value) VALUES ('rows', 0);
345"""
347_INSERT = (
348 "INSERT OR REPLACE INTO profiles (id, method, path, route, grp,"
349 " query_string, search_path, status_code, duration_ms, query_ms,"
350 " query_count, duplicate_count, error_count, blocking_count,"
351 " problem_count, recorded_at, client, problems, payload)"
352 " VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"
353)
356class SQLiteStorage(BaseStorage):
357 """Profiles in a SQLite file.
359 Why you would want it: history survives a restart, and every `uvicorn`
360 worker writes to the same file, so the viewer shows *all* of your traffic
361 rather than the fraction that happened to hit the worker you are talking
362 to. That single fact is what makes the profiler trustworthy under the
363 default multi-worker deployment.
365 **Writes go through a background thread.** `add()` is called from the
366 middleware while the event loop is running, and a synchronous insert there
367 means an `fsync` -- and, with several workers on one WAL file, a lock wait
368 that `sqlite3` implements as a blocking sleep -- happening *on the loop*.
369 Measured at up to 800 ms of stall in a single `add()`. Enqueueing instead
370 keeps the request path at a few microseconds, and batching the drain makes
371 the writes themselves cheaper. Reads flush first, so you never see a stale
372 page.
374 Pass ``background=False`` for synchronous writes (simpler in tests, and
375 fine for a single-worker development server).
376 """
378 def __init__(
379 self,
380 path: str | Path = "profiler.db",
381 max_requests: int = 5000,
382 *,
383 background: bool = True,
384 batch_size: int = 64,
385 linger_seconds: float = 0.02,
386 read_only: bool = False,
387 ) -> None:
388 self.path = str(path)
389 self.read_only = read_only
390 self.max_requests = max_requests
391 self.batch_size = max(1, batch_size)
392 self.linger_seconds = max(0.0, linger_seconds)
393 self._lock = threading.RLock()
394 self._closed = False
396 if read_only:
397 uri = f"file:{urllib.parse.quote(self.path)}?mode=ro"
398 self._conn = sqlite3.connect(uri, uri=True, check_same_thread=False)
399 else:
400 self._conn = sqlite3.connect(self.path, check_same_thread=False)
401 self._conn.row_factory = sqlite3.Row
402 try:
403 with self._lock:
404 self._conn.create_function("py_lower", 1, _lower, deterministic=True)
405 if read_only:
406 self._check_schema()
407 else:
408 # WAL so several worker processes can write concurrently.
409 self._conn.execute("PRAGMA journal_mode=WAL")
410 self._conn.execute("PRAGMA synchronous=NORMAL")
411 self._prepare_schema()
412 except BaseException:
413 # An __init__ that raises leaves no object for the caller to close,
414 # so the connection would leak -- visible as a ResourceWarning
415 # under `-W error` and as a held file handle on Windows.
416 self._conn.close()
417 self._closed = True
418 raise
420 self._queue: queue.Queue[Profile | None] | None = None
421 self._writer: threading.Thread | None = None
422 if background:
423 self._queue = queue.Queue()
424 self._writer = threading.Thread(
425 target=self._drain_forever,
426 name="asgi-profiler-writer",
427 daemon=True,
428 )
429 self._writer.start()
430 # A daemon thread is killed at interpreter exit with whatever is
431 # still queued. Nothing calls close() for a store handed to
432 # install(), so without this the tail of every capture is lost --
433 # which for "profile in staging, read the file later" is the part
434 # you were waiting for.
435 _OPEN_STORES.add(self)
437 # -- schema -----------------------------------------------------------
438 def _check_schema(self) -> None:
439 """Read-only: report an incompatible file, never rewrite it.
441 `_prepare_schema` drops and recreates on a version mismatch, which is
442 the right call for a live capture and catastrophic for the offline
443 viewer -- opening someone's staging capture to look at it must not be
444 able to delete it.
445 """
446 version = self._conn.execute("PRAGMA user_version").fetchone()[0]
447 if version != SCHEMA_VERSION:
448 raise IncompatibleCapture(
449 f"{self.path} was written by a different version of "
450 f"asgi-profiler (schema {version}, expected "
451 f"{SCHEMA_VERSION}). Re-capture with this version."
452 )
454 def _prepare_schema(self) -> None:
455 version = self._conn.execute("PRAGMA user_version").fetchone()[0]
456 existing = self._conn.execute(
457 "SELECT name FROM sqlite_master WHERE type='table' AND name='profiles'"
458 ).fetchone()
459 if existing and version != SCHEMA_VERSION:
460 # An old file from a previous release. Profiling history is
461 # disposable by nature, so rebuild rather than write a migration
462 # for data nobody would miss.
463 self._conn.executescript(
464 "DROP TABLE IF EXISTS profiles; DROP TABLE IF EXISTS statements;"
465 )
466 self._conn.executescript(_SCHEMA)
467 self._conn.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
468 self._conn.execute(
469 "INSERT OR REPLACE INTO meta (key, value) VALUES"
470 " ('rows', (SELECT COUNT(*) FROM profiles))"
471 )
472 self._conn.commit()
474 # -- writing ----------------------------------------------------------
475 def add(self, profile: Profile) -> None:
476 if self._closed or self.read_only:
477 return
478 if self._queue is not None:
479 self._queue.put(profile)
480 return
481 with self._lock:
482 self._write([profile])
484 def flush(self, timeout: float = 30.0) -> None:
485 """Block until every queued profile has been written.
487 Bounded: `queue.join()` alone waits forever, and this is called from
488 the viewer's request handlers, so a wedged writer would take the
489 application down rather than showing a stale page.
490 """
491 if self._queue is None or self._closed:
492 return
493 deadline = time.monotonic() + timeout
494 # `unfinished_tasks` is undocumented but stable since 2.5 -- reading it
495 # is what lets this poll with a deadline instead of `join()`, which
496 # cannot time out.
497 while self._queue.unfinished_tasks:
498 if time.monotonic() > deadline or not self._alive(): 498 ↛ 499line 498 didn't jump to line 499 because the condition on line 498 was never true
499 logger.warning("Timed out flushing profiles to %s", self.path)
500 return
501 time.sleep(0.001)
503 def _alive(self) -> bool:
504 return self._writer is None or self._writer.is_alive()
506 def _drain_forever(self) -> None: # pragma: no cover - thread body
507 """Drain the queue until the sentinel arrives.
509 Every exit path has to call `task_done()` for each item taken, and no
510 exception may escape: `flush()` is a `queue.join()` with no timeout, so
511 a writer that dies with items outstanding parks every future flush --
512 and every viewer request -- forever.
513 """
514 # Not an `assert`: `python -O` strips those, and this one is the
515 # narrowing the rest of the body relies on. A real check costs one
516 # comparison once per thread.
517 if self._queue is None: # pragma: no cover - unreachable by construction
518 return
519 while True:
520 batch: Profiles = []
521 stop = False
522 first = self._queue.get()
523 if first is None:
524 stop = True
525 else:
526 batch.append(first)
527 # Linger briefly to coalesce. Without it the writer wakes for
528 # every single profile whenever it is keeping up, and each wake
529 # is its own transaction -- which puts the producer back in
530 # contention with SQLite work and throws away the whole point
531 # of writing off the request path.
532 deadline = time.monotonic() + self.linger_seconds
533 while len(batch) < self.batch_size:
534 remaining = deadline - time.monotonic()
535 if remaining <= 0:
536 break
537 try:
538 nxt = self._queue.get(timeout=remaining)
539 except queue.Empty:
540 break
541 if nxt is None:
542 stop = True
543 self._queue.task_done()
544 break
545 batch.append(nxt)
547 taken = len(batch)
548 try:
549 if batch:
550 with self._lock:
551 self._write(batch)
552 except Exception:
553 # Losing a batch is bad; losing the writer is worse. Say so --
554 # a diagnostic tool that under-reports traffic in silence is
555 # the one failure nobody would think to check for.
556 logger.warning(
557 "Could not write %d profile(s) to %s",
558 len(batch),
559 self.path,
560 exc_info=True,
561 )
562 finally:
563 for _ in range(taken):
564 self._queue.task_done()
565 if first is None:
566 self._queue.task_done()
567 if stop:
568 return
570 def _write(self, profiles: Profiles) -> None:
571 # Last write wins within a batch, so the row-count bookkeeping below
572 # matches what SQLite will actually store.
573 deduped = {p.id: p for p in profiles}
574 already = {
575 r["id"]
576 for r in self._conn.execute(
577 f"SELECT id FROM profiles WHERE id IN ({','.join('?' * len(deduped))})", # nosec hardcoded_sql_expressions
578 list(deduped),
579 )
580 }
581 rows = [(_row_values(p), p) for p in deduped.values()]
582 if already:
583 # INSERT OR REPLACE allocates a *new* seq, so the replaced
584 # profile's statement rows survive it -- unreachable, but still
585 # aggregated, inflating every count on the statements page.
586 self._conn.execute(
587 "DELETE FROM statements WHERE profile_seq IN" # nosec hardcoded_sql_expressions
588 f" (SELECT seq FROM profiles WHERE id IN"
589 f" ({','.join('?' * len(already))}))",
590 list(already),
591 )
592 self._conn.executemany(_INSERT, [values for values, _ in rows])
593 seqs = {
594 r["id"]: r["seq"]
595 for r in self._conn.execute(
596 "SELECT id, seq FROM profiles WHERE id IN" # nosec hardcoded_sql_expressions
597 f" ({','.join('?' * len(rows))})",
598 [p.id for _, p in rows],
599 )
600 }
601 self._conn.executemany(
602 "INSERT INTO statements"
603 " (profile_seq, sql, sql_hash, grp, method, duration_ms, failed)"
604 " VALUES (?,?,?,?,?,?,?)",
605 [
606 (
607 seqs[p.id],
608 q.sql,
609 q.sql_hash,
610 p.group,
611 p.method,
612 q.duration_ms,
613 1 if q.failed else 0,
614 )
615 for _, p in rows
616 if p.id in seqs
617 for q in p.queries
618 ],
619 )
621 # An INSERT OR REPLACE over an existing id does not add a row. Adding
622 # one to the count anyway makes the trim evict a row too many, and the
623 # history shrinks a little every time an id repeats.
624 # An INSERT OR REPLACE over an existing id does not add a row. Adding
625 # one to the count anyway makes the trim evict a row too many, and the
626 # history shrinks a little every time an id repeats.
627 self._bump_rows(len(deduped) - len(already))
628 self._trim()
629 self._conn.commit()
631 def _bump_rows(self, delta: int) -> int:
632 if delta:
633 self._conn.execute(
634 "UPDATE meta SET value = MAX(0, value + ?) WHERE key = 'rows'",
635 (delta,),
636 )
637 row = self._conn.execute("SELECT value FROM meta WHERE key = 'rows'").fetchone()
638 return int(row["value"]) if row else 0
640 def _trim(self) -> None:
641 """Drop the oldest rows above the cap, in O(overshoot).
643 The obvious `DELETE ... WHERE seq < (SELECT MIN(seq) FROM (... ORDER BY
644 seq DESC LIMIT n))` walks the whole retention window on every insert --
645 390 us per request at the default cap of 5000, paid on the event loop.
646 This walks only as far as the number of rows actually being evicted,
647 which in the steady state is one per insert, while keeping the cap
648 exact: a soft cap that overshoots by a batch is a worse trade than a
649 cheap exact one.
651 The count it works from lives in the file rather than in this process.
652 A per-process counter goes stale the moment another worker calls
653 `clear()`, and a stale-high counter makes every subsequent insert
654 delete the row it just wrote -- a worker that silently stops recording
655 anything at all.
656 """
657 excess = self._bump_rows(0) - self.max_requests
658 if excess <= 0:
659 return
660 row = self._conn.execute(
661 "SELECT seq FROM profiles ORDER BY seq ASC LIMIT 1 OFFSET ?",
662 (excess - 1,),
663 ).fetchone()
664 if row is None: # pragma: no cover - another process trimmed first
665 return
666 cutoff = row["seq"]
667 self._conn.execute("DELETE FROM statements WHERE profile_seq <= ?", (cutoff,))
668 deleted = self._conn.execute("DELETE FROM profiles WHERE seq <= ?", (cutoff,)).rowcount
669 self._bump_rows(-max(0, deleted))
671 def clear(self) -> None:
672 if self.read_only:
673 return
674 self.flush()
675 with self._lock:
676 self._conn.execute("DELETE FROM profiles")
677 self._conn.execute("DELETE FROM statements")
678 self._conn.execute("UPDATE meta SET value = 0 WHERE key = 'rows'")
679 self._conn.commit()
681 def close(self) -> None:
682 """Flush pending writes, stop the writer thread and close the file."""
683 if self._closed: 683 ↛ 684line 683 didn't jump to line 684 because the condition on line 683 was never true
684 return
685 _OPEN_STORES.discard(self)
686 if self._queue is not None:
687 self._queue.put(None)
688 if self._writer is not None: 688 ↛ 690line 688 didn't jump to line 690 because the condition on line 688 was always true
689 self._writer.join(timeout=10)
690 self._closed = True
691 with self._lock:
692 self._conn.close()
694 def __enter__(self) -> SQLiteStorage:
695 return self
697 def __exit__(self, *exc: object) -> None:
698 self.close()
700 # -- reading ----------------------------------------------------------
701 def count(self) -> int:
702 self.flush()
703 with self._lock:
704 row = self._conn.execute("SELECT COUNT(*) AS n FROM profiles").fetchone()
705 return int(row["n"])
707 def get(self, profile_id: str) -> Profile | None:
708 self.flush()
709 with self._lock:
710 row = self._conn.execute(
711 "SELECT * FROM profiles WHERE id = ?", (profile_id,)
712 ).fetchone()
713 return _row_to_profile(row, with_queries=True) if row else None
715 def list(self, *, limit: int | None = None, offset: int = 0) -> Profiles:
716 self.flush()
717 sql = "SELECT * FROM profiles ORDER BY seq DESC LIMIT ? OFFSET ?"
718 with self._lock:
719 rows = self._conn.execute(sql, (-1 if limit is None else limit, offset)).fetchall()
720 return [_row_to_profile(r, with_queries=True) for r in rows]
722 def search(self, filters: Filters, *, page: int = 1, size: int = 50) -> Page:
723 self.flush()
724 where, params = _where(filters)
725 with self._lock:
726 total = int(
727 self._conn.execute(
728 f"SELECT COUNT(*) AS n FROM profiles {where}", # nosec hardcoded_sql_expressions
729 params,
730 ).fetchone()["n"]
731 )
732 size = max(1, size)
733 pages = max(1, -(-total // size))
734 number = min(max(1, page), pages)
735 # `seq DESC` as a tiebreaker everywhere: without it, equal
736 # durations page in an arbitrary order here and newest-first in
737 # MemoryStorage, so the same data reads differently per backend.
738 order = {
739 "recent": "seq DESC",
740 "slowest": "duration_ms DESC, seq DESC",
741 "queries": "query_count DESC, seq DESC",
742 "sql": "query_ms DESC, seq DESC",
743 }.get(filters.order, "seq DESC")
744 rows = self._conn.execute(
745 f"SELECT * FROM profiles {where} ORDER BY {order} LIMIT ? OFFSET ?", # nosec hardcoded_sql_expressions
746 (*params, size, (number - 1) * size),
747 ).fetchall()
748 # The listing shows counters, never individual statements, so skip the
749 # JSON payload entirely -- that is most of the row.
750 return Page(
751 items=[_row_to_profile(r, with_queries=False) for r in rows],
752 total=total,
753 number=number,
754 size=size,
755 )
757 def summarise(self) -> PathSummaries:
758 self.flush()
759 with self._lock:
760 rows = self._conn.execute(
761 "SELECT method, grp, duration_ms, query_count, duplicate_count,"
762 " error_count, problem_count FROM profiles"
763 ).fetchall()
764 # Aggregated in Python rather than SQL because percentiles need the
765 # whole distribution and SQLite has no PERCENTILE_CONT. One narrow
766 # column scan is cheaper than a correlated subquery per group.
767 buckets: dict[tuple[str, str], PathSummary] = {}
768 durations: dict[tuple[str, str], list[float]] = defaultdict(list)
769 for row in rows:
770 key = (row["method"], row["grp"])
771 summary = buckets.get(key)
772 if summary is None:
773 summary = buckets[key] = PathSummary(path=row["grp"], method=key[0])
774 duration = float(row["duration_ms"])
775 summary.count += 1
776 summary.total_ms += duration
777 summary.max_ms = max(summary.max_ms, duration)
778 summary.total_queries += int(row["query_count"])
779 summary.total_duplicates += int(row["duplicate_count"])
780 summary.total_errors += int(row["error_count"])
781 summary.total_problems += int(row["problem_count"])
782 durations[key].append(duration)
783 for key, summary in buckets.items():
784 summary.set_percentiles(durations[key])
785 return sorted(buckets.values(), key=lambda r: r.total_ms, reverse=True)
787 def statements(self, limit: int = 100) -> StatementSummaries:
788 self.flush()
789 with self._lock:
790 rows = self._conn.execute(
791 "SELECT MIN(sql) AS sql, sql_hash, COUNT(*) AS n,"
792 " SUM(duration_ms) AS total,"
793 " MAX(duration_ms) AS worst, SUM(failed) AS failures,"
794 " COUNT(DISTINCT profile_seq) AS requests,"
795 " COUNT(DISTINCT grp) AS routes,"
796 " MIN(grp) AS a_route, MIN(method) AS a_method"
797 " FROM statements GROUP BY sql_hash"
798 # `sql` as a tiebreaker so that a LIMIT returns the same rows
799 # here as in MemoryStorage. Equal totals are common with
800 # coarse timers, and a limit that depends on the backend is
801 # not an interchangeable API.
802 " ORDER BY total DESC, sql ASC LIMIT ?",
803 (max(0, limit),),
804 ).fetchall()
805 return [
806 StatementSummary(
807 sql=r["sql"],
808 sql_hash=r["sql_hash"],
809 count=int(r["n"]),
810 total_ms=float(r["total"] or 0.0),
811 max_ms=float(r["worst"] or 0.0),
812 requests=int(r["requests"]),
813 route_count=int(r["routes"]),
814 sample_route=r["a_route"] or "",
815 sample_method=r["a_method"] or "GET",
816 failures=int(r["failures"] or 0),
817 )
818 for r in rows
819 ]
822def _lower(value: Any) -> Any:
823 return value.lower() if isinstance(value, str) else value
826def _search_text(profile: Profile) -> str:
827 return profile.full_path.lower()
830def _row_values(profile: Profile) -> tuple[Any, ...]:
831 payload = json.dumps(
832 {
833 "request_headers": profile.request_headers,
834 "response_headers": profile.response_headers,
835 "queries": [
836 {
837 "sql": q.sql,
838 "params": q.params,
839 "duration_ms": q.duration_ms,
840 "started_ms": q.started_ms,
841 "blocking": q.blocking,
842 "stack": q.stack,
843 "is_duplicate": q.is_duplicate,
844 "error": q.error,
845 }
846 for q in profile.queries
847 ],
848 }
849 )
850 return (
851 profile.id,
852 profile.method,
853 profile.path,
854 profile.route,
855 profile.group,
856 profile.query_string,
857 _search_text(profile),
858 profile.status_code,
859 profile.duration_ms,
860 profile.query_ms,
861 profile.query_count,
862 profile.duplicate_count,
863 profile.error_count,
864 profile.blocking_count,
865 len(profile.problems),
866 profile.recorded_at.isoformat(),
867 profile.client,
868 json.dumps([problem.as_dict() for problem in profile.problems]),
869 payload,
870 )
873def _where(filters: Filters) -> tuple[str, tuple[Any, ...]]:
874 clauses: list[str] = []
875 params: list[Any] = []
876 if filters.q:
877 # Matched against a column folded at insert time. Calling py_lower per
878 # row turned this into a Python callback over a full table scan --
879 # 16 ms at 20k rows against 3.7 ms for the pure-Python backend, which
880 # is the opposite of the point of pushing the filter down.
881 clauses.append("search_path LIKE ? ESCAPE '\\'")
882 needle = filters.q.lower().replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
883 params.append(f"%{needle}%")
884 if filters.method:
885 clauses.append("method = ?")
886 params.append(filters.method)
887 if filters.route:
888 clauses.append("grp = ?")
889 params.append(filters.route)
890 if filters.status == "err":
891 clauses.append("status_code >= 500")
892 elif filters.status == "warn":
893 clauses.append("status_code >= 400 AND status_code < 500")
894 elif filters.status == "ok":
895 clauses.append("status_code >= 200 AND status_code < 300")
896 if filters.only_problems:
897 clauses.append("problem_count > 0")
898 if filters.only_duplicates:
899 clauses.append("duplicate_count > 0")
900 if filters.only_errors:
901 clauses.append("error_count > 0")
902 if filters.min_ms is not None:
903 clauses.append("duration_ms >= ?")
904 params.append(filters.min_ms)
905 return ("WHERE " + " AND ".join(clauses)) if clauses else "", tuple(params)
908def _row_to_profile(row: sqlite3.Row, *, with_queries: bool) -> Profile:
909 payload = json.loads(row["payload"]) if with_queries else {}
910 profile = Profile(
911 id=row["id"],
912 method=row["method"],
913 path=row["path"],
914 route=row["route"],
915 query_string=row["query_string"],
916 status_code=int(row["status_code"]),
917 duration_ms=float(row["duration_ms"]),
918 recorded_at=_parse_datetime(row["recorded_at"]),
919 request_headers=payload.get("request_headers", {}),
920 response_headers=payload.get("response_headers", {}),
921 client=row["client"],
922 queries=[
923 Query(
924 sql=q["sql"],
925 params=q["params"],
926 duration_ms=q["duration_ms"],
927 started_ms=q.get("started_ms", 0.0),
928 blocking=q.get("blocking", False),
929 stack=q.get("stack", []),
930 is_duplicate=q.get("is_duplicate", False),
931 error=q.get("error"),
932 )
933 for q in payload.get("queries", [])
934 ],
935 # Read whether or not the statements were: this is what puts a badge
936 # on a listing row, and the listing never loads the payload.
937 problems=[Problem.from_dict(d) for d in json.loads(row["problems"] or "[]")],
938 )
939 # Read the counters back rather than recomputing them, so a listing row
940 # stays correct even though its statements were not loaded.
941 profile.query_count = int(row["query_count"])
942 profile.query_ms = float(row["query_ms"])
943 profile.duplicate_count = int(row["duplicate_count"])
944 profile.error_count = int(row["error_count"])
945 profile.blocking_count = int(row["blocking_count"])
946 return profile
949def _parse_datetime(text: str) -> datetime:
950 try:
951 parsed = datetime.fromisoformat(text)
952 except ValueError: # pragma: no cover - defensive
953 return datetime.now(tz=timezone.utc)
954 return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
957def summarise(profiles: Iterable[Profile]) -> PathSummaries:
958 """Group profiles for the summary page.
960 Keyed by the route *pattern* where one is known, so `/users/1` and
961 `/users/2` land on one row instead of two. Grouping by the literal path
962 turns the summary into a deduplicated request log, which is exactly what
963 the page is not for.
964 """
965 rows: dict[tuple[str, str], PathSummary] = {}
966 durations: dict[tuple[str, str], list[float]] = defaultdict(list)
967 for profile in profiles:
968 key = (profile.method, profile.group)
969 row = rows.get(key)
970 if row is None:
971 row = rows[key] = PathSummary(path=profile.group, method=profile.method)
972 row.count += 1
973 row.total_ms += profile.duration_ms
974 row.max_ms = max(row.max_ms, profile.duration_ms)
975 row.total_queries += profile.query_count
976 row.total_duplicates += profile.duplicate_count
977 row.total_errors += profile.error_count
978 row.total_problems += len(profile.problems)
979 durations[key].append(profile.duration_ms)
980 for key, row in rows.items():
981 row.set_percentiles(durations[key])
982 return sorted(rows.values(), key=lambda r: r.total_ms, reverse=True)
985def aggregate_statements(profiles: Iterable[Profile], limit: int = 100) -> StatementSummaries:
986 """Group every recorded statement by its SQL, across all requests.
988 The per-request view answers "why is *this* endpoint slow". This answers
989 "what is costing me across the whole application" -- the statement that
990 runs three times on eleven different endpoints is invisible to the former
991 and obvious here.
992 """
993 rows: dict[str, StatementSummary] = {}
994 routes: dict[str, set[str]] = defaultdict(set)
995 methods: dict[str, set[str]] = defaultdict(set)
996 requests: dict[str, set[str]] = defaultdict(set)
997 for profile in profiles:
998 for query in profile.queries:
999 key = query.sql_hash
1000 row = rows.get(key)
1001 if row is None:
1002 row = rows[key] = StatementSummary(sql=query.sql, sql_hash=key)
1003 row.count += 1
1004 row.total_ms += query.duration_ms
1005 row.max_ms = max(row.max_ms, query.duration_ms)
1006 # `min` throughout, never "the first one seen": SQLite groups with
1007 # MIN(), and a sample that depends on insertion order is not the
1008 # same API. The displayed text is one real execution of the group,
1009 # chosen the same way in both backends.
1010 row.sql = min(row.sql, query.sql)
1011 if query.failed:
1012 row.failures += 1
1013 routes[key].add(profile.group)
1014 methods[key].add(profile.method)
1015 requests[key].add(profile.id)
1016 for key, row in rows.items():
1017 row.route_count = len(routes[key])
1018 row.requests = len(requests[key])
1019 row.sample_route = min(routes[key]) if routes[key] else ""
1020 row.sample_method = min(methods[key]) if methods[key] else "GET"
1021 ordered = sorted(rows.values(), key=lambda r: (-r.total_ms, r.sql))
1022 return ordered[: max(0, limit)]