Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/management_endpoints/common_daily_activity.py: 65%
512 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1import asyncio
2from collections.abc import Awaitable, Callable, Mapping, Sequence
3from collections.abc import Set as AbstractSet
4from datetime import datetime, timedelta, timezone
5from types import MappingProxyType, SimpleNamespace
6from typing import TYPE_CHECKING, Final, Protocol
8from fastapi import HTTPException, status
9from typing_extensions import ReadOnly, TypedDict
11from litellm._logging import verbose_proxy_logger
12from litellm.constants import PTU_SENTINEL_API_KEY
13from litellm.proxy._types import CommonProxyErrors
14from litellm.proxy.spend_tracking.key_metadata_recovery import (
15 attach_user_details,
16 recover_cli_session_key_metadata,
17 recover_double_hashed_key_metadata,
18 recover_key_metadata_from_spend_logs,
19)
20from litellm.proxy.spend_tracking.ptu_feature_flag import is_ptu_cost_attribution_enabled
21from litellm.proxy.utils import PrismaClient
22from litellm.repositories.prisma_protocols import TableActions
23from litellm.repositories.table_repositories import DeletedVerificationTokenRepository
24from litellm.repositories.verification_token_repository import (
25 VerificationTokenRepository,
26)
27from litellm.types.proxy.management_endpoints.common_daily_activity import (
28 BreakdownMetrics,
29 DailySpendData,
30 DailySpendMetadata,
31 GroupedData,
32 KeyMetadata,
33 KeyMetricWithMetadata,
34 MetricWithMetadata,
35 SpendAnalyticsPaginatedResponse,
36 SpendMetrics,
37)
39if TYPE_CHECKING: 39 ↛ 40line 39 didn't jump to line 40 because the condition on line 39 was never true
40 from prisma.models import (
41 LiteLLM_DeletedVerificationToken as PrismaDeletedVerificationToken,
42 )
43 from prisma.models import (
44 LiteLLM_VerificationToken as PrismaVerificationToken,
45 )
47# Mapping from Prisma accessor names to actual PostgreSQL table names.
48_PRISMA_TO_PG_TABLE: Final[Mapping[str, str]] = {
49 "litellm_dailyuserspend": "LiteLLM_DailyUserSpend",
50 "litellm_dailyteamspend": "LiteLLM_DailyTeamSpend",
51 "litellm_dailyorganizationspend": "LiteLLM_DailyOrganizationSpend",
52 "litellm_dailyenduserspend": "LiteLLM_DailyEndUserSpend",
53 "litellm_dailyagentspend": "LiteLLM_DailyAgentSpend",
54 "litellm_dailytagspend": "LiteLLM_DailyTagSpend",
55}
58class DailySpendRecord(Protocol):
59 @property
60 def date(self) -> str: ... 60 ↛ exitline 60 didn't return from function 'date' because
62 @property
63 def api_key(self) -> str: ... 63 ↛ exitline 63 didn't return from function 'api_key' because
65 @property
66 def model(self) -> str | None: ... 66 ↛ exitline 66 didn't return from function 'model' because
68 @property
69 def model_group(self) -> str | None: ... 69 ↛ exitline 69 didn't return from function 'model_group' because
71 @property
72 def custom_llm_provider(self) -> str | None: ... 72 ↛ exitline 72 didn't return from function 'custom_llm_provider' because
74 @property
75 def mcp_namespaced_tool_name(self) -> str | None: ... 75 ↛ exitline 75 didn't return from function 'mcp_namespaced_tool_name' because
77 @property
78 def endpoint(self) -> str | None: ... 78 ↛ exitline 78 didn't return from function 'endpoint' because
80 @property
81 def prompt_tokens(self) -> int: ... 81 ↛ exitline 81 didn't return from function 'prompt_tokens' because
83 @property
84 def completion_tokens(self) -> int: ... 84 ↛ exitline 84 didn't return from function 'completion_tokens' because
86 @property
87 def spend(self) -> float: ... 87 ↛ exitline 87 didn't return from function 'spend' because
89 @property
90 def cache_read_input_tokens(self) -> int: ... 90 ↛ exitline 90 didn't return from function 'cache_read_input_tokens' because
92 @property
93 def cache_creation_input_tokens(self) -> int: ... 93 ↛ exitline 93 didn't return from function 'cache_creation_input_tokens' because
95 @property
96 def compression_saved_tokens(self) -> int: ... 96 ↛ exitline 96 didn't return from function 'compression_saved_tokens' because
98 @property
99 def compression_savings_spend(self) -> float: ... 99 ↛ exitline 99 didn't return from function 'compression_savings_spend' because
101 @property
102 def prompt_caching_savings_spend(self) -> float: ... 102 ↛ exitline 102 didn't return from function 'prompt_caching_savings_spend' because
104 @property
105 def gateway_injected_caching_savings_spend(self) -> float: ... 105 ↛ exitline 105 didn't return from function 'gateway_injected_caching_savings_spend' because
107 @property
108 def autorouter_savings_spend(self) -> float: ... 108 ↛ exitline 108 didn't return from function 'autorouter_savings_spend' because
110 @property
111 def api_requests(self) -> int: ... 111 ↛ exitline 111 didn't return from function 'api_requests' because
113 @property
114 def successful_requests(self) -> int: ... 114 ↛ exitline 114 didn't return from function 'successful_requests' because
116 @property
117 def failed_requests(self) -> int: ... 117 ↛ exitline 117 didn't return from function 'failed_requests' because
119 @property
120 def total_response_time_ms(self) -> int: ... 120 ↛ exitline 120 didn't return from function 'total_response_time_ms' because
122 @property
123 def timed_requests(self) -> int: ... 123 ↛ exitline 123 didn't return from function 'timed_requests' because
126class _KeyMetadataDict(TypedDict, total=False):
127 key_alias: ReadOnly[str | None]
128 team_id: ReadOnly[str | None]
129 user_id: ReadOnly[str | None]
130 user_email: ReadOnly[str | None]
131 key_exists: ReadOnly[bool]
134def _key_metadata(api_key_metadata: Mapping[str, _KeyMetadataDict], api_key: str) -> KeyMetadata:
135 meta: Final = api_key_metadata.get(api_key, {})
136 return KeyMetadata(
137 key_alias=meta.get("key_alias"),
138 team_id=meta.get("team_id"),
139 user_id=meta.get("user_id"),
140 user_email=meta.get("user_email"),
141 key_exists=meta.get("key_exists", False),
142 )
145_WhereValue = str | dict[str, object]
148class _AggregatedSpendData(TypedDict):
149 results: list[DailySpendData]
150 totals: SpendMetrics
153class _GroupingSetsRow(SimpleNamespace):
154 date: str
155 api_key: str | None
156 model: str | None
157 model_group: str | None
158 custom_llm_provider: str | None
159 mcp_namespaced_tool_name: str | None
160 endpoint: str | None
161 group_level: int
162 spend: float | None
163 prompt_tokens: int | None
164 completion_tokens: int | None
165 cache_read_input_tokens: int | None
166 cache_creation_input_tokens: int | None
167 compression_saved_tokens: int | None
168 compression_savings_spend: float | None
169 prompt_caching_savings_spend: float | None
170 gateway_injected_caching_savings_spend: float | None
171 autorouter_savings_spend: float | None
172 api_requests: int | None
173 successful_requests: int | None
174 failed_requests: int | None
175 total_response_time_ms: int | None
176 timed_requests: int | None
179class _EntityRollupRow(_GroupingSetsRow):
180 entity_id: str | None
181 api_key_rolled: int
184def _reported_flat_cost(record: DailySpendRecord | _GroupingSetsRow) -> float:
185 """Flat cost a daily row reports, which is zero unless PTU cost attribution is enabled.
187 Both read paths funnel through here: the paginated path reads the ``ptu_flat_cost``
188 column straight off the row, and the aggregated path reads the SUM() alias. Rows an
189 operator accrued during an earlier opt-in stay in the table, so the gate lives on the
190 read rather than on the query that produced the rows.
192 The row is checked before the flag because this runs once per metric accumulation, and
193 a record fans out across roughly a dozen breakdowns. The flag reads through the secret
194 manager, uncached, so consulting it for every accumulation put thousands of lookups on
195 a shared endpoint that made none before. Only a row actually carrying flat cost, which
196 is a sentinel row, reaches it now.
197 """
198 raw: Final = getattr(record, "ptu_flat_cost", None) or 0.0
199 if not raw: 199 ↛ 201line 199 didn't jump to line 201 because the condition on line 199 was always true
200 return 0.0
201 if not is_ptu_cost_attribution_enabled():
202 return 0.0
203 return raw
206def update_metrics(existing_metrics: SpendMetrics, record: DailySpendRecord) -> SpendMetrics:
207 """Update metrics with new record data.
209 Rollup rows can carry None for numeric fields when SUM() spans zero rows
210 (e.g. a key with no spend), so coalesce to 0 before accumulating to avoid
211 a TypeError. Mirrors the handling in ``_record_to_spend_metrics``.
212 """
213 prompt_tokens: Final = record.prompt_tokens or 0
214 completion_tokens: Final = record.completion_tokens or 0
215 existing_metrics.spend += record.spend or 0.0
216 existing_metrics.flat_cost += _reported_flat_cost(record)
217 existing_metrics.prompt_tokens += prompt_tokens
218 existing_metrics.completion_tokens += completion_tokens
219 existing_metrics.total_tokens += prompt_tokens + completion_tokens
220 existing_metrics.cache_read_input_tokens += record.cache_read_input_tokens or 0
221 existing_metrics.cache_creation_input_tokens += record.cache_creation_input_tokens or 0
222 existing_metrics.compression_saved_tokens += record.compression_saved_tokens or 0
223 existing_metrics.compression_savings_spend += record.compression_savings_spend or 0
224 existing_metrics.prompt_caching_savings_spend += record.prompt_caching_savings_spend or 0
225 existing_metrics.gateway_injected_caching_savings_spend += ( # rebind-ok: this accumulator mutates its target in place for every metric on the row
226 record.gateway_injected_caching_savings_spend or 0
227 )
228 existing_metrics.autorouter_savings_spend += record.autorouter_savings_spend or 0
229 existing_metrics.api_requests += record.api_requests or 0
230 existing_metrics.successful_requests += record.successful_requests or 0
231 existing_metrics.failed_requests += record.failed_requests or 0
232 existing_metrics.total_response_time_ms += record.total_response_time_ms or 0
233 existing_metrics.timed_requests += record.timed_requests or 0
234 return existing_metrics
237def _is_user_agent_tag(tag: str | None) -> bool:
238 """Determine whether a tag should be treated as a User-Agent tag."""
239 if not tag:
240 return False
241 normalized_tag: Final = tag.strip().lower()
242 return normalized_tag.startswith("user-agent:") or normalized_tag.startswith("user agent:")
245def compute_tag_metadata_totals(records: Sequence[DailySpendRecord]) -> SpendMetrics:
246 """
247 Deduplicate spend metrics for tags using request_id, ignoring User-Agent prefixed tags.
249 Each unique request_id contributes at most one record (the tag with max spend) to metadata.
250 """
251 deduped_records: Final[dict[str, DailySpendRecord]] = {}
252 for record in records:
253 request_id: str | None = getattr(record, "request_id", None)
254 if not request_id:
255 continue
257 tag_value = getattr(record, "tag", None)
258 if _is_user_agent_tag(tag_value):
259 continue
261 current_best = deduped_records.get(request_id)
262 if current_best is None or record.spend > current_best.spend:
263 deduped_records[request_id] = record
265 metadata_metrics: Final = SpendMetrics()
266 for record in deduped_records.values():
267 update_metrics(metadata_metrics, record)
268 return metadata_metrics
271def _entity_metadata(
272 entity_metadata_field: Mapping[str, dict[str, object]] | None,
273 entity_id: str,
274) -> dict[str, object]:
275 """The metadata payload for one entity breakdown bucket, empty when the caller passed none."""
276 stored: Final = entity_metadata_field.get(entity_id) if entity_metadata_field else None
277 return stored if stored is not None else {} # mutable-ok: payload pydantic validates into its own dict
280def update_breakdown_metrics(
281 breakdown: BreakdownMetrics,
282 record: DailySpendRecord,
283 model_metadata: Mapping[str, dict[str, object]],
284 provider_metadata: Mapping[str, dict[str, object]],
285 api_key_metadata: Mapping[str, _KeyMetadataDict],
286 entity_id_field: str | None = None,
287 entity_metadata_field: Mapping[str, dict[str, object]] | None = None,
288) -> BreakdownMetrics:
289 """Updates breakdown metrics for a single record using the existing update_metrics function.
291 PTU sentinel rows (api_key == PTU_SENTINEL_API_KEY) add their flat cost to every
292 parent bucket but never appear as an api_key row, and are kept out of the
293 per-request provider breakdown."""
295 is_ptu_sentinel: Final = record.api_key == PTU_SENTINEL_API_KEY
297 # A PTU sentinel row keys on the deployment id so a rename cannot move it, and carries
298 # the operator-facing name in model_group. The breakdown key is rendered directly as a
299 # label, so display the name; two deployments sharing one name merge here, which is
300 # what the write path used to do by collapsing them into a single row.
301 model_key: Final = (record.model_group or record.model) if is_ptu_sentinel else record.model
303 # Update model breakdown
304 if model_key and model_key not in breakdown.models:
305 breakdown.models[model_key] = MetricWithMetadata(
306 metrics=SpendMetrics(),
307 metadata=model_metadata.get(model_key, {}), # Add any model-specific metadata here
308 )
309 if model_key:
310 breakdown.models[model_key].metrics = update_metrics(breakdown.models[model_key].metrics, record)
312 if not is_ptu_sentinel: 312 ↛ 325line 312 didn't jump to line 325 because the condition on line 312 was always true
313 # Update API key breakdown for this model
314 if record.api_key not in breakdown.models[model_key].api_key_breakdown:
315 breakdown.models[model_key].api_key_breakdown[record.api_key] = KeyMetricWithMetadata(
316 metrics=SpendMetrics(),
317 metadata=_key_metadata(api_key_metadata, record.api_key),
318 )
319 breakdown.models[model_key].api_key_breakdown[record.api_key].metrics = update_metrics(
320 breakdown.models[model_key].api_key_breakdown[record.api_key].metrics,
321 record,
322 )
324 # Update model group breakdown
325 model_group_key: Final = record.model_group or record.model
326 if model_group_key and model_group_key not in breakdown.model_groups:
327 breakdown.model_groups[model_group_key] = MetricWithMetadata(
328 metrics=SpendMetrics(),
329 metadata=model_metadata.get(model_group_key, {}),
330 )
331 if model_group_key:
332 breakdown.model_groups[model_group_key].metrics = update_metrics(
333 breakdown.model_groups[model_group_key].metrics, record
334 )
336 if not is_ptu_sentinel: 336 ↛ 348line 336 didn't jump to line 348 because the condition on line 336 was always true
337 # Update API key breakdown for this model
338 if record.api_key not in breakdown.model_groups[model_group_key].api_key_breakdown:
339 breakdown.model_groups[model_group_key].api_key_breakdown[record.api_key] = KeyMetricWithMetadata(
340 metrics=SpendMetrics(),
341 metadata=_key_metadata(api_key_metadata, record.api_key),
342 )
343 breakdown.model_groups[model_group_key].api_key_breakdown[record.api_key].metrics = update_metrics(
344 breakdown.model_groups[model_group_key].api_key_breakdown[record.api_key].metrics,
345 record,
346 )
348 if record.mcp_namespaced_tool_name: 348 ↛ 349line 348 didn't jump to line 349 because the condition on line 348 was never true
349 if record.mcp_namespaced_tool_name not in breakdown.mcp_servers:
350 breakdown.mcp_servers[record.mcp_namespaced_tool_name] = MetricWithMetadata(
351 metrics=SpendMetrics(),
352 metadata={},
353 )
354 breakdown.mcp_servers[record.mcp_namespaced_tool_name].metrics = update_metrics(
355 breakdown.mcp_servers[record.mcp_namespaced_tool_name].metrics, record
356 )
358 # Update API key breakdown for this MCP server
359 if record.api_key not in breakdown.mcp_servers[record.mcp_namespaced_tool_name].api_key_breakdown:
360 breakdown.mcp_servers[record.mcp_namespaced_tool_name].api_key_breakdown[record.api_key] = (
361 KeyMetricWithMetadata(
362 metrics=SpendMetrics(),
363 metadata=_key_metadata(api_key_metadata, record.api_key),
364 )
365 )
367 breakdown.mcp_servers[record.mcp_namespaced_tool_name].api_key_breakdown[
368 record.api_key
369 ].metrics = update_metrics(
370 breakdown.mcp_servers[record.mcp_namespaced_tool_name].api_key_breakdown[record.api_key].metrics,
371 record,
372 )
374 if not is_ptu_sentinel: 374 ↛ 396line 374 didn't jump to line 396 because the condition on line 374 was always true
375 # Update provider breakdown
376 provider: Final = record.custom_llm_provider or "unknown"
377 if provider not in breakdown.providers:
378 breakdown.providers[provider] = MetricWithMetadata(
379 metrics=SpendMetrics(),
380 metadata=provider_metadata.get(provider, {}), # Add any provider-specific metadata here
381 )
382 breakdown.providers[provider].metrics = update_metrics(breakdown.providers[provider].metrics, record)
384 # Update API key breakdown for this provider
385 if record.api_key not in breakdown.providers[provider].api_key_breakdown:
386 breakdown.providers[provider].api_key_breakdown[record.api_key] = KeyMetricWithMetadata(
387 metrics=SpendMetrics(),
388 metadata=_key_metadata(api_key_metadata, record.api_key),
389 )
390 breakdown.providers[provider].api_key_breakdown[record.api_key].metrics = update_metrics(
391 breakdown.providers[provider].api_key_breakdown[record.api_key].metrics,
392 record,
393 )
395 # Update endpoint breakdown
396 if record.endpoint:
397 if record.endpoint not in breakdown.endpoints:
398 breakdown.endpoints[record.endpoint] = MetricWithMetadata(
399 metrics=SpendMetrics(),
400 metadata={},
401 )
402 breakdown.endpoints[record.endpoint].metrics = update_metrics(
403 breakdown.endpoints[record.endpoint].metrics, record
404 )
406 # Update API key breakdown for this endpoint
407 if record.api_key not in breakdown.endpoints[record.endpoint].api_key_breakdown:
408 breakdown.endpoints[record.endpoint].api_key_breakdown[record.api_key] = KeyMetricWithMetadata(
409 metrics=SpendMetrics(),
410 metadata=_key_metadata(api_key_metadata, record.api_key),
411 )
412 breakdown.endpoints[record.endpoint].api_key_breakdown[record.api_key].metrics = update_metrics(
413 breakdown.endpoints[record.endpoint].api_key_breakdown[record.api_key].metrics,
414 record,
415 )
417 if not is_ptu_sentinel: 417 ↛ 427line 417 didn't jump to line 427 because the condition on line 417 was always true
418 # Update api key breakdown
419 if record.api_key not in breakdown.api_keys:
420 breakdown.api_keys[record.api_key] = KeyMetricWithMetadata(
421 metrics=SpendMetrics(),
422 metadata=_key_metadata(api_key_metadata, record.api_key),
423 )
424 breakdown.api_keys[record.api_key].metrics = update_metrics(breakdown.api_keys[record.api_key].metrics, record)
426 # Update entity-specific metrics if entity_id_field is provided
427 if entity_id_field: 427 ↛ 449line 427 didn't jump to line 449 because the condition on line 427 was always true
428 entity_value = getattr(record, entity_id_field, None)
429 entity_value = entity_value if entity_value else "Unassigned" # allow for null entity_id_field
430 if entity_value not in breakdown.entities:
431 breakdown.entities[entity_value] = MetricWithMetadata(
432 metrics=SpendMetrics(),
433 metadata=_entity_metadata(entity_metadata_field, entity_value),
434 )
435 breakdown.entities[entity_value].metrics = update_metrics(breakdown.entities[entity_value].metrics, record)
437 if not is_ptu_sentinel: 437 ↛ 449line 437 didn't jump to line 449 because the condition on line 437 was always true
438 # Update API key breakdown for this entity
439 if record.api_key not in breakdown.entities[entity_value].api_key_breakdown:
440 breakdown.entities[entity_value].api_key_breakdown[record.api_key] = KeyMetricWithMetadata(
441 metrics=SpendMetrics(),
442 metadata=_key_metadata(api_key_metadata, record.api_key),
443 )
444 breakdown.entities[entity_value].api_key_breakdown[record.api_key].metrics = update_metrics(
445 breakdown.entities[entity_value].api_key_breakdown[record.api_key].metrics,
446 record,
447 )
449 return breakdown
452def _spend_logs_window(dates: AbstractSet[str | None]) -> tuple[datetime, datetime] | None:
453 parsed: Final = sorted(day for day in (_parse_spend_date(raw) for raw in dates) if day is not None)
454 if not parsed: 454 ↛ 455line 454 didn't jump to line 455 because the condition on line 454 was never true
455 return None
456 return (parsed[0] - timedelta(days=1), parsed[-1] + timedelta(days=2))
459def _parse_spend_date(raw: str | None) -> datetime | None:
460 if not isinstance(raw, str): 460 ↛ 461line 460 didn't jump to line 461 because the condition on line 460 was never true
461 return None
462 try:
463 return datetime.fromisoformat(raw)
464 except ValueError:
465 return None
468_EMPTY_KEY_METADATA: Final[Mapping[str, _KeyMetadataDict]] = MappingProxyType({})
471async def get_api_key_metadata(
472 prisma_client: PrismaClient,
473 api_keys: AbstractSet[str],
474 spend_logs_window: tuple[datetime, datetime] | None = None,
475) -> Mapping[str, _KeyMetadataDict]:
476 """Get api key metadata, falling back to deleted keys table for keys not found in active table.
478 This ensures that key_alias and team_id are preserved in historical activity logs
479 even after a key is deleted or regenerated. Also recovers aliases for api_key
480 values that were double-hashed by the v1.99 spend-log provenance gate.
481 """
482 key_records: Sequence[PrismaVerificationToken] = await VerificationTokenRepository(prisma_client).table.find_many(
483 where={"token": {"in": list(api_keys)}}
484 )
485 result: Final[dict[str, _KeyMetadataDict]] = {
486 k.token: {
487 "key_alias": k.key_alias,
488 "team_id": k.team_id,
489 "user_id": getattr(k, "user_id", None),
490 "key_exists": True,
491 }
492 for k in key_records
493 }
495 # For any keys not found in the active table, check the deleted keys table
496 missing_keys: Final = api_keys - set(result.keys())
497 if missing_keys: 497 ↛ 520line 497 didn't jump to line 520 because the condition on line 497 was always true
498 try:
499 deleted_key_records: Final[
500 Sequence[PrismaDeletedVerificationToken]
501 ] = await DeletedVerificationTokenRepository(prisma_client).table.find_many(
502 where={"token": {"in": list(missing_keys)}},
503 order={"deleted_at": "desc"},
504 )
505 # Use the most recent deleted record for each token (ordered by deleted_at desc)
506 for k in deleted_key_records: 506 ↛ 507line 506 didn't jump to line 507 because the loop on line 506 never started
507 if k.token not in result:
508 result[k.token] = {
509 "key_alias": k.key_alias,
510 "team_id": k.team_id,
511 "user_id": getattr(k, "user_id", None),
512 }
513 except Exception as e:
514 verbose_proxy_logger.warning(
515 "Failed to fetch deleted key metadata for %d missing keys: %s",
516 len(missing_keys),
517 e,
518 )
520 from_session_keys: Final = await recover_cli_session_key_metadata(prisma_client, api_keys - frozenset(result))
521 still_missing: Final = api_keys - frozenset(result) - frozenset(from_session_keys)
522 from_reverse_hash: Final = (
523 await recover_double_hashed_key_metadata(prisma_client, still_missing) if still_missing else _EMPTY_KEY_METADATA
524 )
525 after_token_recovery: Final = MappingProxyType({**result, **from_session_keys, **from_reverse_hash})
526 unresolved: Final = api_keys - frozenset(after_token_recovery)
527 from_spend_logs: Final = (
528 await recover_key_metadata_from_spend_logs(prisma_client, unresolved, spend_logs_window)
529 if unresolved and spend_logs_window is not None
530 else _EMPTY_KEY_METADATA
531 )
532 combined: Final = MappingProxyType({**after_token_recovery, **from_spend_logs})
533 return await attach_user_details(prisma_client, combined)
536def _adjust_dates_for_timezone(
537 start_date: str,
538 end_date: str,
539 timezone_offset_minutes: int | None,
540 include_current_utc_day: bool = False,
541 utc_now: datetime | None = None,
542) -> tuple[str, str]:
543 """
544 Map a caller-local date range onto UTC bucket keys, extending only the live end.
546 The aggregation table (e.g. LiteLLM_DailyUserSpend) stores spend in whole-UTC-day
547 buckets keyed on date as YYYY-MM-DD. Any conversion of an interior local-day
548 boundary using only date arithmetic must round to whole UTC days, allowing up to
549 24h of slop at each boundary. A previous implementation expanded the SQL range by
550 an extra full UTC day on whichever side the offset pointed, which pulled in 24h of
551 unrelated bucket data per boundary and produced approximately 100% over-counting on
552 single-day queries (e.g. IST May 29 returning UTC May 28 + UTC May 29 in full).
553 Sums of single-day queries then exceeded the equivalent multi-day aggregate, which
554 is mathematically impossible. Historical dates therefore stay a pass-through: the
555 local date is the UTC bucket key, trading boundary slop for monotonic, additive
556 results. Hour-level buckets or pro-rata weighting would fix that properly; both
557 require data the current schema does not store.
559 The end boundary is different when the range reaches the caller's current day. A
560 caller west of UTC asking for a range ending "today" is asking for data up to now,
561 but once UTC has rolled past their local midnight, everything they sent since then
562 sits in the next UTC bucket, which the pass-through excludes: a PT dashboard goes
563 stale every evening from 5pm until local midnight, showing $0 for anything that
564 only started accruing that evening. Extending such a range to today's UTC bucket
565 cannot over-count, because the only part of that bucket outside the caller's range
566 is the future, and the future is empty. ``timezone_offset_minutes`` follows the
567 JS ``Date.getTimezoneOffset`` convention: UTC minus local, positive west of UTC.
569 The extension is strictly opt-in via ``include_current_utc_day`` so a consumer
570 whose axis or reconciliation expects the range to stop at the requested end date
571 keeps today's byte-for-byte behaviour; the cost optimization dashboard opts in.
572 """
573 if not include_current_utc_day or timezone_offset_minutes is None:
574 return start_date, end_date
575 now: Final = utc_now if utc_now is not None else datetime.now(timezone.utc)
576 caller_local_today: Final = (now - timedelta(minutes=timezone_offset_minutes)).date().isoformat()
577 if end_date < caller_local_today:
578 return start_date, end_date
579 return start_date, max(end_date, now.date().isoformat())
582def _build_where_conditions(
583 *,
584 entity_id_field: str,
585 entity_id: str | list[str] | None,
586 start_date: str,
587 end_date: str,
588 model: str | None,
589 api_key: str | list[str] | None,
590 exclude_entity_ids: list[str] | None = None,
591 timezone_offset_minutes: int | None = None,
592 include_current_utc_day: bool = False,
593) -> dict[str, "_WhereValue"]:
594 """Build prisma where clause for daily activity queries."""
595 # Adjust dates for timezone if provided
596 adjusted_start, adjusted_end = _adjust_dates_for_timezone(
597 start_date, end_date, timezone_offset_minutes, include_current_utc_day
598 )
600 where_conditions: Final[dict[str, _WhereValue]] = {
601 "date": {
602 "gte": adjusted_start,
603 "lte": adjusted_end,
604 }
605 }
607 if model:
608 where_conditions["model"] = model
609 if api_key:
610 if isinstance(api_key, list): 610 ↛ 611line 610 didn't jump to line 611 because the condition on line 610 was never true
611 where_conditions["api_key"] = {"in": api_key}
612 else:
613 where_conditions["api_key"] = api_key
615 if entity_id is not None:
616 if isinstance(entity_id, list):
617 where_conditions[entity_id_field] = {"in": entity_id}
618 else:
619 where_conditions[entity_id_field] = {"equals": entity_id}
621 if exclude_entity_ids:
622 current: _WhereValue = where_conditions.get(entity_id_field, {})
623 if isinstance(current, str): 623 ↛ 624line 623 didn't jump to line 624 because the condition on line 623 was never true
624 current = {"equals": current}
625 current["not"] = {"in": exclude_entity_ids}
626 where_conditions[entity_id_field] = current
628 return where_conditions
631def _build_aggregated_where_clause(
632 *,
633 entity_id_field: str,
634 entity_id: str | list[str] | None,
635 adjusted_start: str,
636 adjusted_end: str,
637 model: str | None,
638 api_key: str | list[str] | None, # mutable-ok: filter union shared with the paginated path
639 exclude_entity_ids: list[str] | None, # mutable-ok: filter union shared with the paginated path
640) -> tuple[str, list[str]]:
641 """Build the WHERE clause and $N params shared by the aggregated queries."""
642 sql_conditions: Final[list[str]] = []
643 sql_params: Final[list[str]] = []
644 p = 1 # parameter index (1-based for PostgreSQL $N placeholders)
646 # Date range (always present)
647 sql_conditions.append(f"date >= ${p}")
648 sql_params.append(adjusted_start)
649 p += 1
651 sql_conditions.append(f"date <= ${p}")
652 sql_params.append(adjusted_end)
653 p += 1
655 # Optional entity filter; an empty list must match nothing, not everything
656 if entity_id is not None:
657 if isinstance(entity_id, list): 657 ↛ 658line 657 didn't jump to line 658 because the condition on line 657 was never true
658 if entity_id:
659 placeholders = ", ".join(f"${p + i}" for i in range(len(entity_id)))
660 sql_conditions.append(f'"{entity_id_field}" IN ({placeholders})')
661 sql_params.extend(entity_id)
662 p += len(entity_id)
663 else:
664 sql_conditions.append("FALSE")
665 else:
666 sql_conditions.append(f'"{entity_id_field}" = ${p}')
667 sql_params.append(entity_id)
668 p += 1
670 # Exclude specific entities
671 if exclude_entity_ids:
672 placeholders = ", ".join(f"${p + i}" for i in range(len(exclude_entity_ids)))
673 sql_conditions.append(f'"{entity_id_field}" NOT IN ({placeholders})')
674 sql_params.extend(exclude_entity_ids)
675 p += len(exclude_entity_ids)
677 # Optional model filter
678 if model:
679 sql_conditions.append(f"model = ${p}")
680 sql_params.append(model)
681 p += 1
683 # Optional api_key filter; an empty list must match nothing, not everything
684 if isinstance(api_key, list): 684 ↛ 685line 684 didn't jump to line 685 because the condition on line 684 was never true
685 if api_key:
686 placeholders = ", ".join(f"${p + i}" for i in range(len(api_key)))
687 sql_conditions.append(f"api_key IN ({placeholders})")
688 sql_params.extend(api_key)
689 p += len(api_key)
690 else:
691 sql_conditions.append("FALSE")
692 elif api_key:
693 sql_conditions.append(f"api_key = ${p}")
694 sql_params.append(api_key)
695 p += 1
697 return " AND ".join(sql_conditions), sql_params
700def _ptu_flat_cost_select(table_name: str) -> str:
701 """Only LiteLLM_DailyTeamSpend carries ptu_flat_cost; other daily tables emit a
702 constant zero so the SpendMetrics.flat_cost response shape stays uniform."""
703 if table_name == "litellm_dailyteamspend":
704 return "SUM(ptu_flat_cost)::float AS ptu_flat_cost"
705 return "0::float AS ptu_flat_cost"
708def _build_aggregated_sql_query(
709 *,
710 table_name: str,
711 entity_id_field: str,
712 entity_id: str | list[str] | None, # mutable-ok: filter union shared with the paginated path
713 start_date: str,
714 end_date: str,
715 model: str | None,
716 api_key: str | list[str] | None, # mutable-ok: filter union shared with the paginated path
717 exclude_entity_ids: list[str] | None = None, # mutable-ok: filter union shared with the paginated path
718 timezone_offset_minutes: int | None = None,
719 include_current_utc_day: bool = False,
720) -> tuple[str, list[str]]: # mutable-ok: SQL text plus its ordered $N params
721 """Build a parameterized SQL GROUP BY query for aggregated daily activity.
723 Groups by (date, api_key, model, model_group, custom_llm_provider,
724 mcp_namespaced_tool_name, endpoint) with SUMs on all metric columns.
725 The entity_id column is intentionally omitted from GROUP BY to collapse
726 rows across entities — this is where the biggest row reduction comes from.
728 Returns:
729 Tuple of (sql_query, params_list) ready for prisma_client.db.query_raw().
730 """
731 pg_table: Final = _PRISMA_TO_PG_TABLE.get(table_name)
732 if pg_table is None: 732 ↛ 733line 732 didn't jump to line 733 because the condition on line 732 was never true
733 raise ValueError(f"Unknown table name: {table_name}")
735 adjusted_start, adjusted_end = _adjust_dates_for_timezone(
736 start_date, end_date, timezone_offset_minutes, include_current_utc_day
737 )
739 where_clause, sql_params = _build_aggregated_where_clause(
740 entity_id_field=entity_id_field,
741 entity_id=entity_id,
742 adjusted_start=adjusted_start,
743 adjusted_end=adjusted_end,
744 model=model,
745 api_key=api_key,
746 exclude_entity_ids=exclude_entity_ids,
747 )
749 # Postgres computes every rollup level the response needs — per-date
750 # totals, per-(date, model), per-(date, model, api_key), per-provider,
751 # etc. — in a single pass via GROUPING SETS. The GROUPING() bitmask
752 # encodes which level a row belongs to so Python can dispatch rows
753 # straight into their buckets without re-summing. The leaf grouping
754 # is omitted on purpose: nothing in the response shape needs it once
755 # all the rollups are present.
756 #
757 # TODO: drop the successful_requests/failed_requests aggregates (and the
758 # total_successful_requests metadata they feed) once the admin UI reads SGR
759 # only from LiteLLM_DailyGatewayRequests. The remaining spend, token and
760 # api_requests rollups are still served from here.
761 sql_query: Final = f"""
762 SELECT
763 date,
764 api_key,
765 model,
766 COALESCE(NULLIF(model_group, ''), model) AS model_group,
767 custom_llm_provider,
768 mcp_namespaced_tool_name,
769 endpoint,
770 GROUPING(date, api_key, model, COALESCE(NULLIF(model_group, ''), model),
771 custom_llm_provider, mcp_namespaced_tool_name,
772 endpoint) AS group_level,
773 SUM(spend)::float AS spend,
774 {_ptu_flat_cost_select(table_name)},
775 SUM(prompt_tokens)::bigint AS prompt_tokens,
776 SUM(completion_tokens)::bigint AS completion_tokens,
777 SUM(cache_read_input_tokens)::bigint AS cache_read_input_tokens,
778 SUM(cache_creation_input_tokens)::bigint AS cache_creation_input_tokens,
779 SUM(compression_saved_tokens)::bigint AS compression_saved_tokens,
780 SUM(compression_savings_spend)::float AS compression_savings_spend,
781 SUM(prompt_caching_savings_spend)::float AS prompt_caching_savings_spend,
782 SUM(gateway_injected_caching_savings_spend)::float AS gateway_injected_caching_savings_spend,
783 SUM(autorouter_savings_spend)::float AS autorouter_savings_spend,
784 SUM(api_requests)::bigint AS api_requests,
785 SUM(successful_requests)::bigint AS successful_requests,
786 SUM(failed_requests)::bigint AS failed_requests,
787 SUM(total_response_time_ms)::bigint AS total_response_time_ms,
788 SUM(timed_requests)::bigint AS timed_requests
789 FROM "{pg_table}"
790 WHERE {where_clause}
791 GROUP BY GROUPING SETS (
792 (date),
793 (date, api_key),
794 (date, model),
795 (date, model, api_key),
796 (date, COALESCE(NULLIF(model_group, ''), model)),
797 (date, COALESCE(NULLIF(model_group, ''), model), api_key),
798 (date, custom_llm_provider),
799 (date, custom_llm_provider, api_key),
800 (date, mcp_namespaced_tool_name),
801 (date, mcp_namespaced_tool_name, api_key),
802 (date, endpoint),
803 (date, endpoint, api_key),
804 ()
805 )
806 """
808 return sql_query, sql_params
811def _build_entity_rollup_sql_query(
812 *,
813 table_name: str,
814 entity_id_field: str,
815 entity_id: str | list[str] | None, # mutable-ok: filter union shared with the paginated path
816 start_date: str,
817 end_date: str,
818 model: str | None,
819 api_key: str | list[str] | None, # mutable-ok: filter union shared with the paginated path
820 exclude_entity_ids: list[str] | None = None, # mutable-ok: filter union shared with the paginated path
821 timezone_offset_minutes: int | None = None,
822 include_current_utc_day: bool = False,
823) -> tuple[str, list[str]]: # mutable-ok: SQL text plus its ordered $N params
824 """Per-entity companion to _build_aggregated_sql_query.
826 Two rollup levels over the same WHERE clause — (date, entity) and
827 (date, entity, api_key) — told apart by GROUPING(api_key): 1 when the
828 api_key column is rolled up, 0 when it is part of the key.
829 """
830 pg_table: Final = _PRISMA_TO_PG_TABLE.get(table_name)
831 if pg_table is None: 831 ↛ 832line 831 didn't jump to line 832 because the condition on line 831 was never true
832 raise ValueError(f"Unknown table name: {table_name}")
834 adjusted_start, adjusted_end = _adjust_dates_for_timezone(
835 start_date, end_date, timezone_offset_minutes, include_current_utc_day
836 )
838 where_clause, sql_params = _build_aggregated_where_clause(
839 entity_id_field=entity_id_field,
840 entity_id=entity_id,
841 adjusted_start=adjusted_start,
842 adjusted_end=adjusted_end,
843 model=model,
844 api_key=api_key,
845 exclude_entity_ids=exclude_entity_ids,
846 )
848 sql_query: Final = f"""
849 SELECT
850 "{entity_id_field}" AS entity_id,
851 date,
852 api_key,
853 GROUPING(api_key) AS api_key_rolled,
854 SUM(spend)::float AS spend,
855 {_ptu_flat_cost_select(table_name)},
856 SUM(prompt_tokens)::bigint AS prompt_tokens,
857 SUM(completion_tokens)::bigint AS completion_tokens,
858 SUM(cache_read_input_tokens)::bigint AS cache_read_input_tokens,
859 SUM(cache_creation_input_tokens)::bigint AS cache_creation_input_tokens,
860 SUM(compression_saved_tokens)::bigint AS compression_saved_tokens,
861 SUM(compression_savings_spend)::float AS compression_savings_spend,
862 SUM(prompt_caching_savings_spend)::float AS prompt_caching_savings_spend,
863 SUM(gateway_injected_caching_savings_spend)::float AS gateway_injected_caching_savings_spend,
864 SUM(autorouter_savings_spend)::float AS autorouter_savings_spend,
865 SUM(api_requests)::bigint AS api_requests,
866 SUM(successful_requests)::bigint AS successful_requests,
867 SUM(failed_requests)::bigint AS failed_requests,
868 SUM(total_response_time_ms)::bigint AS total_response_time_ms,
869 SUM(timed_requests)::bigint AS timed_requests
870 FROM "{pg_table}"
871 WHERE {where_clause}
872 GROUP BY GROUPING SETS (
873 (date, "{entity_id_field}"),
874 (date, "{entity_id_field}", api_key)
875 )
876 """
878 return sql_query, sql_params
881def _aggregate_spend_records_sync(
882 *,
883 records: Sequence[DailySpendRecord],
884 api_key_metadata: Mapping[str, _KeyMetadataDict],
885 entity_id_field: str | None,
886 entity_metadata_field: Mapping[str, dict[str, object]] | None,
887) -> _AggregatedSpendData:
888 model_metadata: Final[dict[str, dict[str, object]]] = {}
889 provider_metadata: Final[dict[str, dict[str, object]]] = {}
891 results: Final[list[DailySpendData]] = []
892 total_metrics = SpendMetrics()
893 grouped_data: Final[dict[str, GroupedData]] = {}
895 for record in records:
896 date_str = record.date
897 if date_str not in grouped_data:
898 grouped_data[date_str] = {
899 "metrics": SpendMetrics(),
900 "breakdown": BreakdownMetrics(),
901 }
903 grouped_data[date_str]["metrics"] = update_metrics(grouped_data[date_str]["metrics"], record)
905 grouped_data[date_str]["breakdown"] = update_breakdown_metrics(
906 grouped_data[date_str]["breakdown"],
907 record,
908 model_metadata,
909 provider_metadata,
910 api_key_metadata,
911 entity_id_field=entity_id_field,
912 entity_metadata_field=entity_metadata_field,
913 )
915 total_metrics = update_metrics(total_metrics, record)
917 for date_str, data in grouped_data.items():
918 results.append(
919 DailySpendData(
920 date=datetime.strptime(date_str, "%Y-%m-%d").date(),
921 metrics=data["metrics"],
922 breakdown=data["breakdown"],
923 )
924 )
926 results.sort(key=lambda x: x.date, reverse=True)
928 return {"results": results, "totals": total_metrics}
931async def _aggregate_spend_records(
932 *,
933 prisma_client: PrismaClient,
934 records: Sequence[DailySpendRecord],
935 entity_id_field: str | None,
936 entity_metadata_field: Mapping[str, dict[str, object]] | None,
937) -> _AggregatedSpendData:
938 """Aggregate rows into DailySpendData list and total metrics.
940 The per-row loop is offloaded to a worker thread via asyncio.to_thread so
941 a large result set doesn't peg the event loop.
942 """
943 api_keys: Final[set[str]] = {
944 record.api_key for record in records if record.api_key and record.api_key != PTU_SENTINEL_API_KEY
945 }
947 api_key_metadata: dict[str, _KeyMetadataDict] = {}
948 if api_keys:
949 api_key_metadata = await get_api_key_metadata(
950 prisma_client, api_keys, _spend_logs_window(frozenset(record.date for record in records))
951 )
953 return await asyncio.to_thread(
954 _aggregate_spend_records_sync,
955 records=records,
956 api_key_metadata=api_key_metadata,
957 entity_id_field=entity_id_field,
958 entity_metadata_field=entity_metadata_field,
959 )
962# GROUPING() bitmask values for each grouping set emitted by
963# _build_aggregated_sql_query. Per Postgres semantics, the rightmost argument
964# is the least-significant bit. Argument order:
965# date, api_key, model, model_group, custom_llm_provider,
966# mcp_namespaced_tool_name, endpoint
967# A bit is 1 when the corresponding column is rolled up (i.e. NOT in the
968# current grouping set's key), 0 when the column is part of the key.
969_GROUP_GRAND_TOTAL: Final = 127 # 0b1111111 — all rolled up
970_GROUP_DATE: Final = 63 # 0b0111111 — only date kept
971_GROUP_DATE_API_KEY: Final = 31 # 0b0011111
972_GROUP_DATE_MODEL: Final = 47 # 0b0101111
973_GROUP_DATE_MODEL_API_KEY: Final = 15 # 0b0001111
974_GROUP_DATE_MODEL_GROUP: Final = 55 # 0b0110111
975_GROUP_DATE_MODEL_GROUP_API_KEY: Final = 23 # 0b0010111
976_GROUP_DATE_PROVIDER: Final = 59 # 0b0111011
977_GROUP_DATE_PROVIDER_API_KEY: Final = 27 # 0b0011011
978_GROUP_DATE_MCP: Final = 61 # 0b0111101
979_GROUP_DATE_MCP_API_KEY: Final = 29 # 0b0011101
980_GROUP_DATE_ENDPOINT: Final = 62 # 0b0111110
981_GROUP_DATE_ENDPOINT_API_KEY: Final = 30 # 0b0011110
984def _record_to_spend_metrics(record: _GroupingSetsRow) -> SpendMetrics:
985 """Build a SpendMetrics directly from one already-aggregated rollup row.
987 SUM() over zero rows is SQL NULL, so rollup rows (notably the grand-total
988 row, which Postgres emits even on an empty match) can carry None values.
989 """
990 prompt_tokens: Final = record.prompt_tokens or 0
991 completion_tokens: Final = record.completion_tokens or 0
992 return SpendMetrics(
993 spend=record.spend or 0.0,
994 flat_cost=_reported_flat_cost(record),
995 prompt_tokens=prompt_tokens,
996 completion_tokens=completion_tokens,
997 total_tokens=prompt_tokens + completion_tokens,
998 cache_read_input_tokens=record.cache_read_input_tokens or 0,
999 cache_creation_input_tokens=record.cache_creation_input_tokens or 0,
1000 compression_saved_tokens=record.compression_saved_tokens or 0,
1001 compression_savings_spend=record.compression_savings_spend or 0,
1002 prompt_caching_savings_spend=record.prompt_caching_savings_spend or 0,
1003 gateway_injected_caching_savings_spend=record.gateway_injected_caching_savings_spend or 0,
1004 autorouter_savings_spend=record.autorouter_savings_spend or 0,
1005 api_requests=record.api_requests or 0,
1006 successful_requests=record.successful_requests or 0,
1007 failed_requests=record.failed_requests or 0,
1008 total_response_time_ms=record.total_response_time_ms or 0,
1009 timed_requests=record.timed_requests or 0,
1010 )
1013def _aggregate_grouping_sets_records_sync(
1014 *,
1015 records: Sequence[_GroupingSetsRow],
1016 api_key_metadata: Mapping[str, _KeyMetadataDict],
1017) -> _AggregatedSpendData:
1018 """Build the response from rollup rows produced by the GROUPING SETS query.
1020 Each row carries a `group_level` bitmask (from Postgres GROUPING()) that
1021 identifies which rollup level it belongs to. We dispatch the row's
1022 pre-aggregated metrics straight into the matching bucket — no per-row
1023 summing in Python and no nested update_metrics calls.
1024 """
1025 total_metrics = SpendMetrics()
1026 grouped_data: Final[dict[str, GroupedData]] = {}
1028 def ensure_date(date_str: str) -> GroupedData:
1029 bucket: GroupedData | None = grouped_data.get(date_str)
1030 if bucket is None:
1031 bucket = {"metrics": SpendMetrics(), "breakdown": BreakdownMetrics()}
1032 grouped_data[date_str] = bucket
1033 return bucket
1035 def assign_metric_with_metadata(target: dict[str, MetricWithMetadata], key: str, metrics: SpendMetrics) -> None:
1036 existing: Final = target.get(key)
1037 if existing is None:
1038 target[key] = MetricWithMetadata(metrics=metrics, metadata={})
1039 else:
1040 existing.metrics = metrics
1042 def assign_api_key_breakdown(
1043 target: dict[str, MetricWithMetadata],
1044 parent_key: str,
1045 api_key: str,
1046 metrics: SpendMetrics,
1047 ) -> None:
1048 parent = target.get(parent_key)
1049 if parent is None:
1050 parent = MetricWithMetadata(metrics=SpendMetrics(), metadata={})
1051 target[parent_key] = parent
1052 parent.api_key_breakdown[api_key] = KeyMetricWithMetadata(
1053 metrics=metrics, metadata=_key_metadata(api_key_metadata, api_key)
1054 )
1056 for record in records:
1057 level = record.group_level
1058 metrics = _record_to_spend_metrics(record)
1059 is_ptu_sentinel = record.api_key == PTU_SENTINEL_API_KEY
1061 if level == _GROUP_GRAND_TOTAL: 1061 ↛ 1065line 1061 didn't jump to line 1065 because the condition on line 1061 was always true
1062 total_metrics = metrics
1063 continue
1065 if level == _GROUP_DATE:
1066 ensure_date(record.date)["metrics"] = metrics
1067 continue
1069 breakdown = ensure_date(record.date)["breakdown"]
1071 if level == _GROUP_DATE_API_KEY:
1072 if record.api_key and not is_ptu_sentinel:
1073 breakdown.api_keys[record.api_key] = KeyMetricWithMetadata(
1074 metrics=metrics,
1075 metadata=_key_metadata(api_key_metadata, record.api_key),
1076 )
1077 elif level == _GROUP_DATE_MODEL:
1078 if record.model:
1079 assign_metric_with_metadata(breakdown.models, record.model, metrics)
1080 elif level == _GROUP_DATE_MODEL_API_KEY:
1081 if record.model and record.api_key and not is_ptu_sentinel:
1082 assign_api_key_breakdown(breakdown.models, record.model, record.api_key, metrics)
1083 elif level == _GROUP_DATE_MODEL_GROUP:
1084 if record.model_group:
1085 assign_metric_with_metadata(breakdown.model_groups, record.model_group, metrics)
1086 elif level == _GROUP_DATE_MODEL_GROUP_API_KEY:
1087 if record.model_group and record.api_key and not is_ptu_sentinel:
1088 assign_api_key_breakdown(
1089 breakdown.model_groups,
1090 record.model_group,
1091 record.api_key,
1092 metrics,
1093 )
1094 elif level == _GROUP_DATE_PROVIDER:
1095 # Only PTU sentinel rows carry ptu_flat_cost and they have no provider, so at
1096 # this level the sentinel's cost would land under "unknown". Withholding the
1097 # flat cost matches the per-row path, which skips sentinel rows outright. The
1098 # bucket itself is still assigned unconditionally: a legacy row predating the
1099 # api_requests column backfills to all zeroes, and skipping those would drop a
1100 # provider the base build reported.
1101 provider_metrics = metrics.model_copy(update={"flat_cost": 0.0}) # mutable-ok: pydantic update payload
1102 provider = record.custom_llm_provider or "unknown"
1103 assign_metric_with_metadata(breakdown.providers, provider, provider_metrics)
1104 elif level == _GROUP_DATE_PROVIDER_API_KEY:
1105 if record.api_key and not is_ptu_sentinel:
1106 provider = record.custom_llm_provider or "unknown"
1107 assign_api_key_breakdown(breakdown.providers, provider, record.api_key, metrics)
1108 elif level == _GROUP_DATE_MCP:
1109 if record.mcp_namespaced_tool_name:
1110 assign_metric_with_metadata(breakdown.mcp_servers, record.mcp_namespaced_tool_name, metrics)
1111 elif level == _GROUP_DATE_MCP_API_KEY:
1112 if record.mcp_namespaced_tool_name and record.api_key:
1113 assign_api_key_breakdown(
1114 breakdown.mcp_servers,
1115 record.mcp_namespaced_tool_name,
1116 record.api_key,
1117 metrics,
1118 )
1119 elif level == _GROUP_DATE_ENDPOINT:
1120 if record.endpoint:
1121 assign_metric_with_metadata(breakdown.endpoints, record.endpoint, metrics)
1122 elif level == _GROUP_DATE_ENDPOINT_API_KEY:
1123 if record.endpoint and record.api_key:
1124 assign_api_key_breakdown(breakdown.endpoints, record.endpoint, record.api_key, metrics)
1126 results: Final = [
1127 DailySpendData(
1128 date=datetime.strptime(date_str, "%Y-%m-%d").date(),
1129 metrics=data["metrics"],
1130 breakdown=data["breakdown"],
1131 )
1132 for date_str, data in grouped_data.items()
1133 ]
1134 results.sort(key=lambda x: x.date, reverse=True)
1136 return {"results": results, "totals": total_metrics}
1139async def _aggregate_grouping_sets_records(
1140 *,
1141 prisma_client: PrismaClient,
1142 records: Sequence[_GroupingSetsRow],
1143) -> _AggregatedSpendData:
1144 """Async wrapper: fetch api_key_metadata, then dispatch on a worker thread."""
1145 api_keys: Final[set[str]] = {r.api_key for r in records if r.api_key and r.api_key != PTU_SENTINEL_API_KEY}
1147 api_key_metadata: dict[str, _KeyMetadataDict] = {}
1148 if api_keys: 1148 ↛ 1149line 1148 didn't jump to line 1149 because the condition on line 1148 was never true
1149 api_key_metadata = await get_api_key_metadata(
1150 prisma_client, api_keys, _spend_logs_window(frozenset(r.date for r in records))
1151 )
1153 return await asyncio.to_thread(
1154 _aggregate_grouping_sets_records_sync,
1155 records=records,
1156 api_key_metadata=api_key_metadata,
1157 )
1160async def get_daily_activity(
1161 prisma_client: PrismaClient | None,
1162 table_name: str,
1163 entity_id_field: str,
1164 entity_id: str | list[str] | None,
1165 entity_metadata_field: Mapping[str, dict[str, object]] | None,
1166 start_date: str | None,
1167 end_date: str | None,
1168 model: str | None,
1169 api_key: str | list[str] | None,
1170 page: int,
1171 page_size: int,
1172 exclude_entity_ids: list[str] | None = None,
1173 metadata_metrics_func: Callable[[Sequence[DailySpendRecord]], SpendMetrics] | None = None,
1174 timezone_offset_minutes: int | None = None,
1175 include_current_utc_day: bool = False,
1176 resolve_entity_metadata: Callable[[Sequence[DailySpendRecord]], Awaitable[dict[str, dict[str, object]]]]
1177 | None = None,
1178) -> SpendAnalyticsPaginatedResponse:
1179 """Common function to get daily activity for any entity type.
1181 ``resolve_entity_metadata`` lets a caller resolve entity metadata from the
1182 rows actually on the page (e.g. user_id -> user_email) instead of fetching
1183 the whole entity table upfront, which matters when the entity set is
1184 unbounded.
1185 """
1187 if prisma_client is None: 1187 ↛ 1188line 1187 didn't jump to line 1188 because the condition on line 1187 was never true
1188 raise HTTPException(
1189 status_code=500,
1190 detail={"error": CommonProxyErrors.db_not_connected_error.value},
1191 )
1193 if start_date is None or end_date is None:
1194 raise HTTPException(
1195 status_code=status.HTTP_400_BAD_REQUEST,
1196 detail={"error": "Please provide start_date and end_date"},
1197 )
1199 try:
1200 where_conditions: Final = _build_where_conditions(
1201 entity_id_field=entity_id_field,
1202 entity_id=entity_id,
1203 start_date=start_date,
1204 end_date=end_date,
1205 model=model,
1206 api_key=api_key,
1207 exclude_entity_ids=exclude_entity_ids,
1208 timezone_offset_minutes=timezone_offset_minutes,
1209 include_current_utc_day=include_current_utc_day,
1210 )
1212 spend_table: Final[TableActions[DailySpendRecord]] = getattr(prisma_client.db, table_name)
1214 # Get total count for pagination
1215 total_count: Final[int] = await spend_table.count(where=where_conditions)
1217 # Fetch paginated results.
1218 # ``date`` alone is not a unique sort key -- a busy tenant has many
1219 # rows per date (one per api_key, model, model_group, provider,
1220 # endpoint, ...), so offset pagination over ``date desc`` lands on
1221 # arbitrary boundaries and the same row can be skipped on one page
1222 # and returned on another. A client that pages through and sums the
1223 # per-page metrics (the Usage dashboard) then gets a non-deterministic
1224 # total. Adding ``id`` (the row's UUID primary key, present on both
1225 # LiteLLM_DailyUserSpend and LiteLLM_DailyTeamSpend) as a tiebreaker
1226 # gives every page a stable cursor (#30164).
1227 daily_spend_data: Final[Sequence[DailySpendRecord]] = await spend_table.find_many(
1228 where=where_conditions,
1229 order=[
1230 {"date": "desc"},
1231 {"id": "asc"},
1232 ],
1233 skip=(page - 1) * page_size,
1234 take=page_size,
1235 )
1237 resolved_entity_metadata = entity_metadata_field
1238 if resolve_entity_metadata is not None:
1239 resolved_entity_metadata = {
1240 **(entity_metadata_field or {}),
1241 **(await resolve_entity_metadata(daily_spend_data)),
1242 }
1244 aggregated: Final = await _aggregate_spend_records(
1245 prisma_client=prisma_client,
1246 records=daily_spend_data,
1247 entity_id_field=entity_id_field,
1248 entity_metadata_field=resolved_entity_metadata,
1249 )
1251 metadata_metrics = aggregated["totals"]
1252 if metadata_metrics_func: 1252 ↛ 1253line 1252 didn't jump to line 1253 because the condition on line 1252 was never true
1253 metadata_metrics = metadata_metrics_func(daily_spend_data)
1255 return SpendAnalyticsPaginatedResponse(
1256 results=aggregated["results"],
1257 metadata=DailySpendMetadata(
1258 total_spend=metadata_metrics.spend,
1259 total_flat_cost=metadata_metrics.flat_cost,
1260 total_prompt_tokens=metadata_metrics.prompt_tokens,
1261 total_completion_tokens=metadata_metrics.completion_tokens,
1262 total_tokens=metadata_metrics.total_tokens,
1263 total_api_requests=metadata_metrics.api_requests,
1264 total_successful_requests=metadata_metrics.successful_requests,
1265 total_failed_requests=metadata_metrics.failed_requests,
1266 total_cache_read_input_tokens=metadata_metrics.cache_read_input_tokens,
1267 total_cache_creation_input_tokens=metadata_metrics.cache_creation_input_tokens,
1268 total_compression_saved_tokens=metadata_metrics.compression_saved_tokens,
1269 total_compression_savings_spend=metadata_metrics.compression_savings_spend,
1270 total_prompt_caching_savings_spend=metadata_metrics.prompt_caching_savings_spend,
1271 total_gateway_injected_caching_savings_spend=metadata_metrics.gateway_injected_caching_savings_spend,
1272 total_autorouter_savings_spend=metadata_metrics.autorouter_savings_spend,
1273 total_response_time_ms=metadata_metrics.total_response_time_ms,
1274 total_timed_requests=metadata_metrics.timed_requests,
1275 page=page,
1276 total_pages=-(-total_count // page_size), # Ceiling division
1277 has_more=(page * page_size) < total_count,
1278 ),
1279 )
1281 except Exception as e:
1282 verbose_proxy_logger.exception("Error fetching daily activity: %s", e)
1283 raise HTTPException(
1284 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
1285 detail={"error": f"Failed to fetch analytics: {e}"},
1286 )
1289def _fold_entity_rollups_sync(
1290 *,
1291 results: Sequence[DailySpendData],
1292 entity_rows: Sequence[_EntityRollupRow],
1293 api_key_metadata: Mapping[str, _KeyMetadataDict],
1294 entity_metadata_field: Mapping[str, dict[str, object]] | None, # mutable-ok: shared field shape
1295) -> None:
1296 """Write breakdown.entities onto the already-built per-day results."""
1297 by_date: Final = {day.date.strftime("%Y-%m-%d"): day for day in results} # mutable-ok: local fold index
1299 for row in entity_rows:
1300 day = by_date.get(row.date)
1301 if day is None:
1302 continue
1304 entities = day.breakdown.entities
1305 entity_id = row.entity_id or "Unassigned"
1306 bucket = entities.get(entity_id)
1307 if bucket is None:
1308 bucket = MetricWithMetadata(
1309 metrics=SpendMetrics(),
1310 metadata=_entity_metadata(entity_metadata_field, entity_id),
1311 )
1312 entities[entity_id] = bucket
1314 metrics = _record_to_spend_metrics(row)
1315 if row.api_key_rolled:
1316 bucket.metrics = metrics
1317 elif row.api_key and row.api_key != PTU_SENTINEL_API_KEY:
1318 bucket.api_key_breakdown[row.api_key] = KeyMetricWithMetadata(
1319 metrics=metrics, metadata=_key_metadata(api_key_metadata, row.api_key)
1320 )
1323async def get_daily_activity_aggregated(
1324 prisma_client: PrismaClient | None,
1325 table_name: str,
1326 entity_id_field: str,
1327 entity_id: str | list[str] | None,
1328 entity_metadata_field: Mapping[str, dict[str, object]] | None,
1329 start_date: str | None,
1330 end_date: str | None,
1331 model: str | None,
1332 api_key: str | list[str] | None, # mutable-ok: filter union shared with the paginated path
1333 exclude_entity_ids: list[str] | None = None,
1334 timezone_offset_minutes: int | None = None,
1335 include_entity_breakdown: bool = False,
1336 include_current_utc_day: bool = False,
1337) -> SpendAnalyticsPaginatedResponse:
1338 """Aggregated variant that returns the full result set (no pagination).
1340 Uses SQL GROUP BY to aggregate rows in the database rather than fetching
1341 all individual rows into Python. This collapses rows across entities
1342 (users/teams/orgs), reducing ~150k rows to ~2-3k grouped rows.
1344 include_entity_breakdown runs a small companion rollup query and folds
1345 `breakdown.entities` onto the response, as entity-scoped views like Team Usage need.
1347 Matches the response model of the paginated endpoint so the UI does not need to transform.
1348 """
1349 if prisma_client is None: 1349 ↛ 1350line 1349 didn't jump to line 1350 because the condition on line 1349 was never true
1350 raise HTTPException(
1351 status_code=500,
1352 detail={"error": CommonProxyErrors.db_not_connected_error.value},
1353 )
1355 if start_date is None or end_date is None: 1355 ↛ 1356line 1355 didn't jump to line 1356 because the condition on line 1355 was never true
1356 raise HTTPException(
1357 status_code=status.HTTP_400_BAD_REQUEST,
1358 detail={"error": "Please provide start_date and end_date"},
1359 )
1361 try:
1362 sql_query, sql_params = _build_aggregated_sql_query(
1363 table_name=table_name,
1364 entity_id_field=entity_id_field,
1365 entity_id=entity_id,
1366 start_date=start_date,
1367 end_date=end_date,
1368 model=model,
1369 api_key=api_key,
1370 exclude_entity_ids=exclude_entity_ids,
1371 timezone_offset_minutes=timezone_offset_minutes,
1372 include_current_utc_day=include_current_utc_day,
1373 )
1375 entity_query: Final = (
1376 _build_entity_rollup_sql_query(
1377 table_name=table_name,
1378 entity_id_field=entity_id_field,
1379 entity_id=entity_id,
1380 start_date=start_date,
1381 end_date=end_date,
1382 model=model,
1383 api_key=api_key,
1384 exclude_entity_ids=exclude_entity_ids,
1385 timezone_offset_minutes=timezone_offset_minutes,
1386 include_current_utc_day=include_current_utc_day,
1387 )
1388 if include_entity_breakdown
1389 else None
1390 )
1392 # Execute the GROUPING SETS query (one row per rollup level), alongside
1393 # the per-entity companion rollup when the caller wants entities.
1394 raw_rows, raw_entity_rows = (
1395 await asyncio.gather(
1396 prisma_client.db.query_raw(sql_query, *sql_params),
1397 prisma_client.db.query_raw(entity_query[0], *entity_query[1]),
1398 )
1399 if entity_query is not None
1400 else (await prisma_client.db.query_raw(sql_query, *sql_params), None)
1401 )
1403 records: Final = [_GroupingSetsRow(**row) for row in (raw_rows or [])]
1405 # The grouping-sets dispatcher places each row directly in its bucket
1406 # using the row's GROUPING() bitmask. No Python-side summing needed.
1407 aggregated: Final = await _aggregate_grouping_sets_records(
1408 prisma_client=prisma_client,
1409 records=records,
1410 )
1412 if raw_entity_rows: 1412 ↛ 1413line 1412 didn't jump to line 1413 because the condition on line 1412 was never true
1413 entity_records: Final = tuple(_EntityRollupRow(**row) for row in raw_entity_rows)
1414 entity_api_keys: Final = frozenset(
1415 r.api_key for r in entity_records if r.api_key and r.api_key != PTU_SENTINEL_API_KEY
1416 )
1417 entity_key_metadata: Final = (
1418 await get_api_key_metadata(
1419 prisma_client, entity_api_keys, _spend_logs_window(frozenset(r.date for r in entity_records))
1420 )
1421 if entity_api_keys
1422 else {} # mutable-ok: matches the helper's dict return
1423 )
1424 await asyncio.to_thread(
1425 _fold_entity_rollups_sync,
1426 results=aggregated["results"],
1427 entity_rows=entity_records,
1428 api_key_metadata=entity_key_metadata,
1429 entity_metadata_field=entity_metadata_field,
1430 )
1432 return SpendAnalyticsPaginatedResponse(
1433 results=aggregated["results"],
1434 metadata=DailySpendMetadata(
1435 total_spend=aggregated["totals"].spend,
1436 total_flat_cost=aggregated["totals"].flat_cost,
1437 total_prompt_tokens=aggregated["totals"].prompt_tokens,
1438 total_completion_tokens=aggregated["totals"].completion_tokens,
1439 total_tokens=aggregated["totals"].total_tokens,
1440 total_api_requests=aggregated["totals"].api_requests,
1441 total_successful_requests=aggregated["totals"].successful_requests,
1442 total_failed_requests=aggregated["totals"].failed_requests,
1443 total_cache_read_input_tokens=aggregated["totals"].cache_read_input_tokens,
1444 total_cache_creation_input_tokens=aggregated["totals"].cache_creation_input_tokens,
1445 total_compression_saved_tokens=aggregated["totals"].compression_saved_tokens,
1446 total_compression_savings_spend=aggregated["totals"].compression_savings_spend,
1447 total_prompt_caching_savings_spend=aggregated["totals"].prompt_caching_savings_spend,
1448 total_gateway_injected_caching_savings_spend=aggregated[
1449 "totals"
1450 ].gateway_injected_caching_savings_spend,
1451 total_autorouter_savings_spend=aggregated["totals"].autorouter_savings_spend,
1452 total_response_time_ms=aggregated["totals"].total_response_time_ms,
1453 total_timed_requests=aggregated["totals"].timed_requests,
1454 page=1,
1455 total_pages=1,
1456 has_more=False,
1457 ),
1458 )
1460 except Exception as e:
1461 verbose_proxy_logger.exception("Error fetching aggregated daily activity: %s", e)
1462 raise HTTPException(
1463 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
1464 detail={"error": f"Failed to fetch analytics: {e}"},
1465 )