Coverage for src/local_deep_research/metrics/query_utils.py: 100%
40 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1"""Common query utilities for metrics module."""
3from datetime import datetime, timedelta, UTC
4from typing import Any, TypedDict
6from sqlalchemy import Column, case, func
7from sqlalchemy.orm import Session
10# Single source of truth for period → days across all metrics endpoints.
11# Covers BOTH UI vocabularies: the main dashboard / context-overflow pages
12# send '3m'/'1y'; the standalone link-analytics page sends '90d'/'365d'.
13# Both alias to the same day counts so every metrics endpoint agrees.
14# 'all' maps to None (no time limit).
15PERIOD_DAYS_MAP = {
16 "7d": 7,
17 "30d": 30,
18 "3m": 90,
19 "90d": 90,
20 "1y": 365,
21 "365d": 365,
22 "all": None,
23}
26def get_period_days(period: str, default: int = 30) -> int | None:
27 """Convert a period string to number of days.
29 Returns None for 'all' (no time limit); unknown values use ``default``.
30 """
31 return PERIOD_DAYS_MAP.get(period, default)
34def get_period_cutoff(period: str) -> datetime | None:
35 """UTC cutoff datetime for a period string, or None for 'all'.
37 Delegates to :func:`get_period_days` so there is exactly one period
38 vocabulary. Unknown values default to 30 days.
39 """
40 days = get_period_days(period)
41 if days is None: # 'all'
42 return None
43 return datetime.now(UTC) - timedelta(days=days)
46def get_time_filter_condition(period: str, timestamp_column: Column) -> Any:
47 """Get SQLAlchemy condition for time filtering.
49 Args:
50 period: Time period ('7d', '30d', '3m', '1y', 'all')
51 timestamp_column: SQLAlchemy timestamp column to filter on
53 Returns:
54 SQLAlchemy condition object or None for 'all'
55 """
56 cutoff = get_period_cutoff(period)
57 if cutoff is None:
58 return None
59 return timestamp_column >= cutoff
62class TruncationSummary(TypedDict):
63 """Aggregated context-overflow stats over a time window.
65 Token-summary fields (total_tokens, prompt/completion sums and avgs,
66 max_prompt_tokens) are computed in the same query because they share
67 the same row set and time window — one scan instead of three.
68 """
70 total_requests: int
71 requests_with_context: int
72 truncated_requests: int
73 truncation_rate: float # raw percentage, unrounded
74 avg_tokens_truncated: float # raw, unrounded
75 total_tokens: int
76 total_prompt_tokens: int
77 total_completion_tokens: int
78 avg_prompt_tokens: float # raw, unrounded
79 avg_completion_tokens: float # raw, unrounded
80 max_prompt_tokens: int
83def get_context_overflow_truncation_summary(
84 session: Session, period: str, research_mode: str = "all"
85) -> TruncationSummary:
86 """Single-source aggregation of truncation + token-summary stats.
88 Both /metrics/api/metrics and /metrics/api/context-overflow surface
89 the same truncation rate; computing it in two places risks the two
90 summaries silently disagreeing for the same time window.
92 Truncation fields and token-summary fields are computed in one merged
93 query (CASE-based AVG isolates truncated rows; SQL AVG ignores NULLs).
94 Returns raw values — callers round/cast for their own display contract.
96 research_mode defaults to "all" (no mode filter). Pass "quick" or
97 "detailed" to scope to that mode — matches the rest of api_metrics()
98 so the dashboard's mode toggle stays in sync with this panel.
99 """
100 from ..database.models import TokenUsage
102 time_condition = get_time_filter_condition(period, TokenUsage.timestamp)
103 mode_condition = get_research_mode_condition(
104 research_mode, TokenUsage.research_mode
105 )
107 base = session.query(TokenUsage)
108 if time_condition is not None:
109 base = base.filter(time_condition)
110 if mode_condition is not None:
111 base = base.filter(mode_condition)
113 row = base.with_entities(
114 func.count(TokenUsage.id).label("total_requests"),
115 func.sum(
116 case((TokenUsage.context_limit.isnot(None), 1), else_=0)
117 ).label("requests_with_context"),
118 func.sum(
119 case((TokenUsage.context_truncated.is_(True), 1), else_=0)
120 ).label("truncated_requests"),
121 func.avg(
122 case(
123 (
124 TokenUsage.context_truncated.is_(True),
125 TokenUsage.tokens_truncated,
126 ),
127 else_=None,
128 )
129 ).label("avg_tokens_truncated"),
130 func.coalesce(func.sum(TokenUsage.total_tokens), 0).label(
131 "total_tokens"
132 ),
133 func.coalesce(func.sum(TokenUsage.prompt_tokens), 0).label(
134 "total_prompt_tokens"
135 ),
136 func.coalesce(func.sum(TokenUsage.completion_tokens), 0).label(
137 "total_completion_tokens"
138 ),
139 func.avg(TokenUsage.prompt_tokens).label("avg_prompt_tokens"),
140 func.avg(TokenUsage.completion_tokens).label("avg_completion_tokens"),
141 func.max(TokenUsage.prompt_tokens).label("max_prompt_tokens"),
142 ).first()
144 total_requests = int(row.total_requests or 0) if row else 0
145 requests_with_context = int(row.requests_with_context or 0) if row else 0
146 truncated_requests = int(row.truncated_requests or 0) if row else 0
148 truncation_rate = (
149 (truncated_requests / requests_with_context) * 100
150 if requests_with_context > 0
151 else 0.0
152 )
154 avg_tokens_truncated = float(row.avg_tokens_truncated or 0) if row else 0.0
156 return TruncationSummary(
157 total_requests=total_requests,
158 requests_with_context=requests_with_context,
159 truncated_requests=truncated_requests,
160 truncation_rate=truncation_rate,
161 avg_tokens_truncated=avg_tokens_truncated,
162 total_tokens=int(row.total_tokens or 0) if row else 0,
163 total_prompt_tokens=int(row.total_prompt_tokens or 0) if row else 0,
164 total_completion_tokens=(
165 int(row.total_completion_tokens or 0) if row else 0
166 ),
167 avg_prompt_tokens=float(row.avg_prompt_tokens or 0) if row else 0.0,
168 avg_completion_tokens=(
169 float(row.avg_completion_tokens or 0) if row else 0.0
170 ),
171 max_prompt_tokens=int(row.max_prompt_tokens or 0) if row else 0,
172 )
175def get_research_mode_condition(research_mode: str, mode_column: Column) -> Any:
176 """Get SQLAlchemy condition for research mode filtering.
178 Args:
179 research_mode: Research mode ('quick', 'detailed', 'all')
180 mode_column: SQLAlchemy column to filter on
182 Returns:
183 SQLAlchemy condition object or None for 'all'
184 """
185 if research_mode == "all":
186 return None
187 if research_mode in ["quick", "detailed"]:
188 return mode_column == research_mode
189 return None