Coverage for src/local_deep_research/metrics/query_utils.py: 100%
43 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-03 23:15 +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
10PERIOD_DAYS_MAP = {"7d": 7, "30d": 30, "90d": 90, "365d": 365, "all": None}
13def get_period_days(period: str, default: int = 30) -> int | None:
14 """Convert a period string to number of days.
16 Returns None for 'all' (no time limit).
17 """
18 return PERIOD_DAYS_MAP.get(period, default)
21def get_time_filter_condition(period: str, timestamp_column: Column) -> Any:
22 """Get SQLAlchemy condition for time filtering.
24 Args:
25 period: Time period ('7d', '30d', '3m', '1y', 'all')
26 timestamp_column: SQLAlchemy timestamp column to filter on
28 Returns:
29 SQLAlchemy condition object or None for 'all'
30 """
31 if period == "all":
32 return None
33 if period == "7d":
34 cutoff = datetime.now(UTC) - timedelta(days=7)
35 elif period == "30d":
36 cutoff = datetime.now(UTC) - timedelta(days=30)
37 elif period == "3m":
38 cutoff = datetime.now(UTC) - timedelta(days=90)
39 elif period == "1y":
40 cutoff = datetime.now(UTC) - timedelta(days=365)
41 else:
42 # Default to 30 days for unknown periods
43 cutoff = datetime.now(UTC) - timedelta(days=30)
45 return timestamp_column >= cutoff
48class TruncationSummary(TypedDict):
49 """Aggregated context-overflow stats over a time window.
51 Token-summary fields (total_tokens, prompt/completion sums and avgs,
52 max_prompt_tokens) are computed in the same query because they share
53 the same row set and time window — one scan instead of three.
54 """
56 total_requests: int
57 requests_with_context: int
58 truncated_requests: int
59 truncation_rate: float # raw percentage, unrounded
60 avg_tokens_truncated: float # raw, unrounded
61 total_tokens: int
62 total_prompt_tokens: int
63 total_completion_tokens: int
64 avg_prompt_tokens: float # raw, unrounded
65 avg_completion_tokens: float # raw, unrounded
66 max_prompt_tokens: int
69def get_context_overflow_truncation_summary(
70 session: Session, period: str, research_mode: str = "all"
71) -> TruncationSummary:
72 """Single-source aggregation of truncation + token-summary stats.
74 Both /metrics/api/metrics and /metrics/api/context-overflow surface
75 the same truncation rate; computing it in two places risks the two
76 summaries silently disagreeing for the same time window.
78 Truncation fields and token-summary fields are computed in one merged
79 query (CASE-based AVG isolates truncated rows; SQL AVG ignores NULLs).
80 Returns raw values — callers round/cast for their own display contract.
82 research_mode defaults to "all" (no mode filter). Pass "quick" or
83 "detailed" to scope to that mode — matches the rest of api_metrics()
84 so the dashboard's mode toggle stays in sync with this panel.
85 """
86 from ..database.models import TokenUsage
88 time_condition = get_time_filter_condition(period, TokenUsage.timestamp)
89 mode_condition = get_research_mode_condition(
90 research_mode, TokenUsage.research_mode
91 )
93 base = session.query(TokenUsage)
94 if time_condition is not None:
95 base = base.filter(time_condition)
96 if mode_condition is not None:
97 base = base.filter(mode_condition)
99 row = base.with_entities(
100 func.count(TokenUsage.id).label("total_requests"),
101 func.sum(
102 case((TokenUsage.context_limit.isnot(None), 1), else_=0)
103 ).label("requests_with_context"),
104 func.sum(
105 case((TokenUsage.context_truncated.is_(True), 1), else_=0)
106 ).label("truncated_requests"),
107 func.avg(
108 case(
109 (
110 TokenUsage.context_truncated.is_(True),
111 TokenUsage.tokens_truncated,
112 ),
113 else_=None,
114 )
115 ).label("avg_tokens_truncated"),
116 func.coalesce(func.sum(TokenUsage.total_tokens), 0).label(
117 "total_tokens"
118 ),
119 func.coalesce(func.sum(TokenUsage.prompt_tokens), 0).label(
120 "total_prompt_tokens"
121 ),
122 func.coalesce(func.sum(TokenUsage.completion_tokens), 0).label(
123 "total_completion_tokens"
124 ),
125 func.avg(TokenUsage.prompt_tokens).label("avg_prompt_tokens"),
126 func.avg(TokenUsage.completion_tokens).label("avg_completion_tokens"),
127 func.max(TokenUsage.prompt_tokens).label("max_prompt_tokens"),
128 ).first()
130 total_requests = int(row.total_requests or 0) if row else 0
131 requests_with_context = int(row.requests_with_context or 0) if row else 0
132 truncated_requests = int(row.truncated_requests or 0) if row else 0
134 truncation_rate = (
135 (truncated_requests / requests_with_context) * 100
136 if requests_with_context > 0
137 else 0.0
138 )
140 avg_tokens_truncated = float(row.avg_tokens_truncated or 0) if row else 0.0
142 return TruncationSummary(
143 total_requests=total_requests,
144 requests_with_context=requests_with_context,
145 truncated_requests=truncated_requests,
146 truncation_rate=truncation_rate,
147 avg_tokens_truncated=avg_tokens_truncated,
148 total_tokens=int(row.total_tokens or 0) if row else 0,
149 total_prompt_tokens=int(row.total_prompt_tokens or 0) if row else 0,
150 total_completion_tokens=(
151 int(row.total_completion_tokens or 0) if row else 0
152 ),
153 avg_prompt_tokens=float(row.avg_prompt_tokens or 0) if row else 0.0,
154 avg_completion_tokens=(
155 float(row.avg_completion_tokens or 0) if row else 0.0
156 ),
157 max_prompt_tokens=int(row.max_prompt_tokens or 0) if row else 0,
158 )
161def get_research_mode_condition(research_mode: str, mode_column: Column) -> Any:
162 """Get SQLAlchemy condition for research mode filtering.
164 Args:
165 research_mode: Research mode ('quick', 'detailed', 'all')
166 mode_column: SQLAlchemy column to filter on
168 Returns:
169 SQLAlchemy condition object or None for 'all'
170 """
171 if research_mode == "all":
172 return None
173 if research_mode in ["quick", "detailed"]:
174 return mode_column == research_mode
175 return None