Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/spend_tracking/spend_tracking_utils.py: 53%
626 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1import json
2import os
3import re
4import secrets
5from collections.abc import Mapping, Sequence
6from datetime import datetime, timezone
7from datetime import datetime as dt
8from functools import reduce
9from types import MappingProxyType
10from typing import TYPE_CHECKING, Final, Literal, Protocol, cast, runtime_checkable
12from pydantic import BaseModel, JsonValue
14import litellm
15from litellm._logging import verbose_proxy_logger
16from litellm.constants import (
17 CLI_SESSION_KEY_PREFIX,
18 EMPTY_MAPPING,
19 LITELLM_PROXY_MASTER_KEY_ALIAS,
20 LITELLM_TRUNCATED_PAYLOAD_FIELD,
21 LITELLM_TRUNCATION_DB_SAFEGUARD_NOTE,
22 LITTELM_CLI_SERVICE_ACCOUNT_NAME,
23 LITTELM_INTERNAL_HEALTH_SERVICE_ACCOUNT_NAME,
24 MAX_SPEND_LOG_MODEL_NAME_LENGTH,
25 MCP_SPEND_LOG_MODEL_PREFIX,
26 REDACTED_BY_LITELM_STRING,
27 SESSION_ID_OMITTED_METADATA_KEY,
28 UNKNOWN_MODEL_SPEND_LOG_MODEL,
29)
30from litellm.constants import (
31 MAX_STRING_LENGTH_PROMPT_IN_DB as DEFAULT_MAX_STRING_LENGTH_PROMPT_IN_DB,
32)
33from litellm.litellm_core_utils.classifier_logging import classifier_audit_fields, without_classifier_audit
34from litellm.litellm_core_utils.core_helpers import (
35 get_litellm_metadata_from_kwargs,
36 reconstruct_model_name,
37)
38from litellm.litellm_core_utils.get_llm_provider_logic import declared_authenticating_provider
39from litellm.litellm_core_utils.internal_call_metadata import is_unbilled_non_inference_call
40from litellm.litellm_core_utils.litellm_logging import (
41 coerce_model_access_groups,
42 is_valid_sha256_hash,
43 request_model_access_groups_from_litellm_params,
44)
45from litellm.litellm_core_utils.ptu_pricing import azure_spillover
46from litellm.litellm_core_utils.safe_json_dumps import safe_dumps, strip_null_bytes
47from litellm.proxy._types import SpendLogsMetadata, SpendLogsPayload, SpendLogsRouterMetadata
48from litellm.proxy.route_llm_request import ProxyModelNotFoundError
49from litellm.proxy.spend_tracking.spend_log_error_logger import spend_log_error
50from litellm.proxy.utils import PrismaClient, hash_token
51from litellm.types.router import DeploymentTypedDict, LiteLLM_Params
52from litellm.types.utils import (
53 PROMPT_CARRYING_GUARDRAIL_FIELDS,
54 AzureSpillover,
55 CallTypes,
56 CostBreakdown,
57 LlmProviders,
58 StandardLoggingGuardrailInformation,
59 StandardLoggingMCPToolCall,
60 StandardLoggingModelInformation,
61 StandardLoggingPayload,
62 StandardLoggingPayloadErrorInformation,
63 StandardLoggingVectorStoreRequest,
64 VectorStoreSearchResponse,
65)
66from litellm.utils import get_end_user_id_for_cost_tracking
68if TYPE_CHECKING: 68 ↛ 69line 68 didn't jump to line 69 because the condition on line 68 was never true
69 from litellm.router import Router
72def _get_max_string_length_prompt_in_db() -> int:
73 """
74 Resolve prompt truncation threshold at runtime so values loaded later via
75 proxy config environment_variables are honored.
76 """
77 max_length_str: Final = os.getenv("MAX_STRING_LENGTH_PROMPT_IN_DB")
78 if max_length_str is None: 78 ↛ 80line 78 didn't jump to line 80 because the condition on line 78 was always true
79 return DEFAULT_MAX_STRING_LENGTH_PROMPT_IN_DB
80 try:
81 return int(max_length_str)
82 except (TypeError, ValueError):
83 return DEFAULT_MAX_STRING_LENGTH_PROMPT_IN_DB
86def _is_master_key(api_key: str | None, _master_key: str | None) -> bool:
87 """
88 Raw-only constant-time master-key comparison. The hashed form is never
89 considered equivalent — only the raw master-key string matches.
90 """
91 if _master_key is None or api_key is None:
92 return False
93 return secrets.compare_digest(api_key, _master_key)
96_HASHED_JWT_RE = re.compile(r"hashed-jwt-[a-fA-F0-9]{64}")
97_NON_SECRET_KEY_ALIASES: Final = frozenset(
98 {
99 LITELLM_PROXY_MASTER_KEY_ALIAS,
100 LITTELM_INTERNAL_HEALTH_SERVICE_ACCOUNT_NAME,
101 LITTELM_CLI_SERVICE_ACCOUNT_NAME,
102 }
103)
106def _is_cli_session_alias(value: str, key_alias: object) -> bool:
107 return value.startswith(f"{CLI_SESSION_KEY_PREFIX}-") and value == key_alias
110def _is_non_secret_key_value(value: str, *, key_alias: object = None) -> bool:
111 return (
112 value in _NON_SECRET_KEY_ALIASES
113 or is_valid_sha256_hash(value)
114 or _HASHED_JWT_RE.fullmatch(value) is not None
115 or _is_cli_session_alias(value, key_alias)
116 )
119def _redact_logged_api_key(
120 value: str | None, *, already_redacted: bool = False, key_alias: object = None
121) -> str | None:
122 if not isinstance(value, str) or not value:
123 return None
124 stripped: Final = re.sub(r"(?i)^bearer ", "", value)
125 if not stripped: 125 ↛ 126line 125 didn't jump to line 126 because the condition on line 125 was never true
126 return None
127 if already_redacted and _is_non_secret_key_value(stripped, key_alias=key_alias):
128 return stripped
129 return hash_token(stripped)
132def _get_router_metadata_for_spend_log(
133 metadata: Mapping[str, object] | None,
134 requested_model: str | None,
135 selected_model: str | None,
136 selected_provider: str | None,
137 router_correlation_id: str | None,
138) -> SpendLogsRouterMetadata | None:
139 model_info: Final = metadata.get("model_info") if metadata is not None else None
140 if not isinstance(model_info, Mapping) or model_info.get("internal_router_model") is not True: 140 ↛ 142line 140 didn't jump to line 142 because the condition on line 140 was always true
141 return None
142 return SpendLogsRouterMetadata(
143 requested_model=requested_model or None,
144 selected_model=selected_model or None,
145 selected_provider=selected_provider or None,
146 router_correlation_id=router_correlation_id,
147 )
150_STAMPED_METADATA_KEYS: Final = frozenset(
151 (
152 "router_metadata",
153 "azure_spillover",
154 "autorouter_savings",
155 "autorouter_savings_estimate",
156 "autorouter_baseline_observation",
157 )
158)
161def _get_spend_logs_metadata(
162 metadata: dict | None,
163 applied_guardrails: list[str] | None = None,
164 batch_models: list[str] | None = None,
165 batch_successful_requests: int | None = None,
166 batch_failed_requests: int | None = None,
167 mcp_tool_call_metadata: StandardLoggingMCPToolCall | None = None,
168 vector_store_request_metadata: list[StandardLoggingVectorStoreRequest] | None = None,
169 guardrail_information: list[StandardLoggingGuardrailInformation] | None = None,
170 usage_object: dict | None = None,
171 model_map_information: StandardLoggingModelInformation | None = None,
172 cold_storage_object_key: str | None = None,
173 litellm_overhead_time_ms: float | None = None,
174 cost_breakdown: CostBreakdown | None = None,
175 litellm_call_id: str | None = None,
176 autorouter_savings: float | None = None,
177 autorouter_savings_estimate: Mapping[str, JsonValue] | None = None,
178 autorouter_baseline_observation: str | None = None,
179 router_metadata: SpendLogsRouterMetadata | None = None,
180 azure_spillover: AzureSpillover | None = None,
181) -> SpendLogsMetadata:
182 if metadata is None: 182 ↛ 183line 182 didn't jump to line 183 because the condition on line 182 was never true
183 return SpendLogsMetadata(
184 user_api_key=None,
185 user_api_key_alias=None,
186 user_api_key_team_id=None,
187 user_api_key_project_id=None,
188 user_api_key_project_alias=None,
189 user_api_key_org_id=None,
190 user_api_key_user_id=None,
191 user_api_key_team_alias=None,
192 spend_logs_metadata=None,
193 requester_ip_address=None,
194 user_agent=None,
195 additional_usage_values=None,
196 applied_guardrails=None,
197 status="success",
198 error_information=None,
199 proxy_server_request=None,
200 batch_models=None,
201 batch_successful_requests=None,
202 batch_failed_requests=None,
203 mcp_tool_call_metadata=None,
204 vector_store_request_metadata=None,
205 model_map_information=None,
206 usage_object=None,
207 guardrail_information=None,
208 internal_call_origin=None,
209 eval_information=None,
210 cold_storage_object_key=cold_storage_object_key,
211 litellm_overhead_time_ms=None,
212 attempted_retries=None,
213 max_retries=None,
214 attempted_fallbacks=None,
215 original_model_group=None,
216 cost_breakdown=None,
217 compression_savings=None,
218 autorouter_savings=autorouter_savings,
219 autorouter_savings_estimate=autorouter_savings_estimate,
220 autorouter_baseline_observation=autorouter_baseline_observation,
221 litellm_gateway_injected_cache=None,
222 litellm_call_id=litellm_call_id,
223 router_metadata=router_metadata,
224 azure_spillover=azure_spillover,
225 )
226 verbose_proxy_logger.debug(
227 "getting payload for SpendLogs, available keys in metadata: " + str(list(metadata.keys()))
228 )
230 # Filter the metadata dictionary to include only the specified keys
231 clean_metadata: Final = SpendLogsMetadata(
232 **MappingProxyType(
233 {key: metadata.get(key) for key in SpendLogsMetadata.__annotations__ if key not in _STAMPED_METADATA_KEYS}
234 ),
235 autorouter_savings=autorouter_savings,
236 autorouter_savings_estimate=autorouter_savings_estimate,
237 autorouter_baseline_observation=autorouter_baseline_observation,
238 router_metadata=router_metadata,
239 azure_spillover=azure_spillover,
240 )
241 _raw_key: Final = clean_metadata.get("user_api_key")
242 _trusted_hash: Final = metadata.get("user_api_key_hash")
243 _key_alias: Final = metadata.get("user_api_key_alias")
244 _already_redacted: Final = (
245 isinstance(_trusted_hash, str)
246 and _is_non_secret_key_value(_trusted_hash, key_alias=_key_alias)
247 and _trusted_hash == _raw_key
248 )
249 clean_metadata["user_api_key"] = _redact_logged_api_key(
250 _raw_key, already_redacted=_already_redacted, key_alias=_key_alias
251 )
252 clean_metadata["applied_guardrails"] = applied_guardrails
253 clean_metadata["batch_models"] = batch_models
254 clean_metadata["batch_successful_requests"] = batch_successful_requests
255 clean_metadata["batch_failed_requests"] = batch_failed_requests
256 clean_metadata["mcp_tool_call_metadata"] = mcp_tool_call_metadata
257 clean_metadata["vector_store_request_metadata"] = _get_vector_store_request_for_spend_logs_payload(
258 vector_store_request_metadata
259 )
260 clean_metadata["guardrail_information"] = _sanitize_guardrail_information_for_spend_logs(guardrail_information)
261 clean_metadata["usage_object"] = usage_object
262 clean_metadata["model_map_information"] = model_map_information
263 clean_metadata["cold_storage_object_key"] = cold_storage_object_key
264 clean_metadata["litellm_overhead_time_ms"] = litellm_overhead_time_ms
265 clean_metadata["cost_breakdown"] = cost_breakdown
266 clean_metadata["litellm_call_id"] = litellm_call_id
268 return clean_metadata
271BATCH_COST_REQUEST_ID_SUFFIX: Final = "_batch_cost"
274def get_spend_logs_id(call_type: str, response_obj: dict, kwargs: dict) -> str | None:
275 standard_logging_payload = kwargs.get("standard_logging_object")
276 candidate_ids: Final = (
277 response_obj.get("id"),
278 standard_logging_payload.get("id") if isinstance(standard_logging_payload, dict) else None,
279 kwargs.get("litellm_call_id"),
280 )
281 resolved_id: Final = next(
282 (candidate for candidate in candidate_ids if isinstance(candidate, str) and candidate), None
283 )
284 if resolved_id is not None and call_type == CallTypes.aretrieve_batch.value:
285 return f"{resolved_id}{BATCH_COST_REQUEST_ID_SUFFIX}"
286 return resolved_id
289_MISSING_ATTRIBUTE: Final = object()
292def _attribute_or_missing(source: object, name: str) -> object:
293 return getattr(source, name, _MISSING_ATTRIBUTE)
296@runtime_checkable
297class _ModelDumpable(Protocol):
298 def model_dump(self) -> object: ... 298 ↛ exitline 298 didn't return from function 'model_dump' because
301def _dumped_usage_info(usage_info: object) -> object:
302 if isinstance(usage_info, _ModelDumpable):
303 return usage_info.model_dump()
304 instance_dict: Final = _attribute_or_missing(usage_info, "__dict__")
305 if instance_dict is not _MISSING_ATTRIBUTE:
306 return instance_dict
307 return usage_info
310def _extract_usage_for_ocr_call(response_obj: object, response_obj_dict: dict) -> dict:
311 """
312 Extract usage information for OCR/AOCR calls.
314 OCR responses use usage_info (with pages_processed) instead of token-based usage.
316 Args:
317 response_obj: The raw response object (can be dict, BaseModel, or other)
318 response_obj_dict: Dictionary representation of the response object
320 Returns:
321 A dict with prompt_tokens=0, completion_tokens=0, total_tokens=0,
322 and pages_processed from usage_info.
323 """
324 usage_info = None
326 # Try to extract usage_info from dict
327 if isinstance(response_obj_dict, dict) and "usage_info" in response_obj_dict: 327 ↛ 328line 327 didn't jump to line 328 because the condition on line 327 was never true
328 usage_info = response_obj_dict.get("usage_info")
330 # Try to extract usage_info from object attributes if not found in dict
331 if not usage_info: 331 ↛ 337line 331 didn't jump to line 337 because the condition on line 331 was always true
332 attribute_usage_info: Final = _attribute_or_missing(response_obj, "usage_info")
333 if attribute_usage_info is not _MISSING_ATTRIBUTE: 333 ↛ 334line 333 didn't jump to line 334 because the condition on line 333 was never true
334 usage_info = _dumped_usage_info(attribute_usage_info)
336 # For OCR, we track pages instead of tokens
337 if usage_info is not None: 337 ↛ 339line 337 didn't jump to line 339 because the condition on line 337 was never true
338 # Handle dict or object with attributes
339 if isinstance(usage_info, dict):
340 result: Final = {
341 "prompt_tokens": 0, # OCR doesn't use traditional tokens
342 "completion_tokens": 0,
343 "total_tokens": 0,
344 }
345 # Add all fields from usage_info, including pages_processed
346 for key, value in usage_info.items():
347 result[key] = value
348 # Ensure pages_processed exists
349 if "pages_processed" not in result:
350 result["pages_processed"] = 0
351 return result
352 else:
353 return {
354 "prompt_tokens": 0,
355 "completion_tokens": 0,
356 "total_tokens": 0,
357 "pages_processed": 0,
358 }
359 else:
360 return {}
363def get_request_model_access_groups(kwargs: Mapping[str, object] | None) -> tuple[str, ...]:
364 """Model access groups that authorized this request, as stamped onto request metadata at auth time."""
365 if kwargs is None: 365 ↛ 366line 365 didn't jump to line 366 because the condition on line 365 was never true
366 return ()
368 standard_logging_payload: Final = kwargs.get("standard_logging_object")
369 if isinstance(standard_logging_payload, Mapping):
370 from_payload: Final = coerce_model_access_groups(standard_logging_payload.get("request_model_access_groups"))
371 if from_payload: 371 ↛ 372line 371 didn't jump to line 372 because the condition on line 371 was never true
372 return from_payload
374 litellm_params: Final = kwargs.get("litellm_params")
375 if not isinstance(litellm_params, Mapping): 375 ↛ 376line 375 didn't jump to line 376 because the condition on line 375 was never true
376 return ()
377 return request_model_access_groups_from_litellm_params(litellm_params)
380def _sl_attribution_fallback(
381 standard_logging_payload: StandardLoggingPayload | None,
382 field: Literal["model_id", "model_group", "api_base", "custom_llm_provider"],
383) -> str:
384 if standard_logging_payload is None:
385 return ""
386 return standard_logging_payload.get(field) or ""
389def _deployment_provider(deployment: DeploymentTypedDict) -> str | None:
390 litellm_params: Final = LiteLLM_Params.model_validate(deployment["litellm_params"])
391 if litellm.LiteLLMProxyChatConfig.should_use_litellm_proxy_by_default(litellm_params=litellm_params):
392 return LlmProviders.LITELLM_PROXY.value
393 declared: Final = declared_authenticating_provider(litellm_params.model, litellm_params.custom_llm_provider)
394 if declared is not None:
395 return declared
396 try:
397 _, provider, _, _ = litellm.get_llm_provider(
398 model=litellm_params.model, custom_llm_provider=litellm_params.custom_llm_provider
399 )
400 except litellm.exceptions.BadRequestError:
401 return None
402 return provider or None
405def _model_group_provider(model_group: str, llm_router: "Router | None") -> str | None:
406 if llm_router is None or not model_group:
407 return None
408 providers: Final = frozenset(
409 provider
410 for deployment in llm_router.get_model_list(model_name=model_group) or ()
411 if (provider := _deployment_provider(deployment)) is not None
412 )
413 return next(iter(providers)) if len(providers) == 1 else None
416def _is_configured_model_group(model_group: str, llm_router: "Router | None") -> bool:
417 if llm_router is None or not model_group: 417 ↛ 419line 417 didn't jump to line 419 because the condition on line 417 was always true
418 return False
419 return llm_router.is_recognized_model(model_group) or model_group in llm_router.team_public_model_names
422def _looks_like_model_name(model: str) -> bool:
423 candidate: Final = model.removeprefix(MCP_SPEND_LOG_MODEL_PREFIX)
424 return len(candidate) <= MAX_SPEND_LOG_MODEL_NAME_LENGTH and not any(char.isspace() for char in candidate)
427_TRUNCATION_MARKER: Final = re.compile(
428 rf"\.\.\. \({re.escape(LITELLM_TRUNCATED_PAYLOAD_FIELD)} skipped \d+ chars\. "
429 rf"{re.escape(LITELLM_TRUNCATION_DB_SAFEGUARD_NOTE)}\) \.\.\."
430)
431_SCRUBBED_ERROR_TEXT_FIELDS: Final = frozenset(("error_message", "traceback"))
434def _raw_model_spellings(raw_model: str) -> tuple[str, ...]:
435 return tuple(dict.fromkeys((raw_model, repr(raw_model)[1:-1], json.dumps(raw_model)[1:-1])))
438def _overlap_at_end(text: str, spelling: str) -> int:
439 lengths: Final = range(min(len(text), len(spelling) - 1), 0, -1)
440 return next((length for length in lengths if text.endswith(spelling[:length])), 0)
443def _overlap_at_start(text: str, spelling: str) -> int:
444 lengths: Final = range(min(len(text), len(spelling) - 1), 0, -1)
445 return next((length for length in lengths if text.startswith(spelling[-length:])), 0)
448def _scrub_raw_model_split_by_truncation(text: str, spellings: tuple[str, ...]) -> str:
449 marker: Final = _TRUNCATION_MARKER.search(text)
450 if marker is None: 450 ↛ 452line 450 didn't jump to line 452 because the condition on line 450 was always true
451 return text
452 head: Final = text[: marker.start()]
453 tail: Final = text[marker.end() :]
454 head_cut: Final = max(_overlap_at_end(head, spelling) for spelling in spellings)
455 tail_cut: Final = max(_overlap_at_start(tail, spelling) for spelling in spellings)
456 return "".join(
457 (
458 head[: len(head) - head_cut],
459 UNKNOWN_MODEL_SPEND_LOG_MODEL if head_cut else "",
460 marker.group(0),
461 UNKNOWN_MODEL_SPEND_LOG_MODEL if tail_cut else "",
462 tail[tail_cut:],
463 )
464 )
467def _scrub_raw_model_from_error_text(text: str, spellings: tuple[str, ...]) -> str:
468 whole_occurrences_scrubbed: Final = reduce(
469 lambda scrubbed, spelling: scrubbed.replace(spelling, UNKNOWN_MODEL_SPEND_LOG_MODEL), spellings, text
470 )
471 return _scrub_raw_model_split_by_truncation(whole_occurrences_scrubbed, spellings)
474def _scrub_raw_model_from_error_information(
475 error_information: StandardLoggingPayloadErrorInformation | None, raw_model: str
476) -> StandardLoggingPayloadErrorInformation | None:
477 if error_information is None or not raw_model:
478 return error_information
479 spellings: Final = _raw_model_spellings(raw_model)
480 return cast(
481 StandardLoggingPayloadErrorInformation,
482 {
483 key: _scrub_raw_model_from_error_text(value, spellings)
484 if key in _SCRUBBED_ERROR_TEXT_FIELDS and isinstance(value, str)
485 else value
486 for key, value in error_information.items()
487 },
488 )
491def get_logging_payload(
492 kwargs: dict | None,
493 response_obj: object,
494 start_time: datetime,
495 end_time: datetime,
496 llm_router: "Router | None" = None,
497) -> SpendLogsPayload:
498 if kwargs is None: 498 ↛ 499line 498 didn't jump to line 499 because the condition on line 498 was never true
499 kwargs = {}
501 rejected_as_unknown_model: Final = isinstance(response_obj, ProxyModelNotFoundError)
502 if response_obj is None: 502 ↛ 503line 502 didn't jump to line 503 because the condition on line 502 was never true
503 response_obj = {}
504 elif not isinstance(response_obj, BaseModel) and not isinstance(response_obj, dict):
505 response_obj = {"result": str(response_obj)}
506 # standardize this function to be used across, s3, dynamoDB, langfuse logging
507 litellm_params: Final = kwargs.get("litellm_params", {})
508 metadata: Final = get_litellm_metadata_from_kwargs(kwargs)
509 completion_start_time: Final = kwargs.get("completion_start_time", end_time)
510 call_type: Final = kwargs.get("call_type")
511 cache_hit: Final = kwargs.get("cache_hit", False)
513 # Convert response_obj to dict first
514 if isinstance(response_obj, dict): 514 ↛ 516line 514 didn't jump to line 516 because the condition on line 514 was always true
515 response_obj_dict = response_obj
516 elif isinstance(response_obj, BaseModel):
517 response_obj_dict = response_obj.model_dump()
518 else:
519 response_obj_dict = {}
521 # Handle OCR responses which use usage_info instead of usage
522 usage: dict = {}
523 if call_type in ["ocr", "aocr"]:
524 usage = _extract_usage_for_ocr_call(response_obj, response_obj_dict)
525 elif not is_unbilled_non_inference_call(call_type, metadata, response_obj_dict):
526 # Use response_obj_dict instead of response_obj to avoid calling .get() on Pydantic models
527 _usage: Final = response_obj_dict.get("usage", None) or {}
528 if isinstance(_usage, litellm.Usage): 528 ↛ 529line 528 didn't jump to line 529 because the condition on line 528 was never true
529 usage = dict(_usage)
530 elif isinstance(_usage, dict): 530 ↛ 536line 530 didn't jump to line 536 because the condition on line 530 was always true
531 usage = _usage
533 # A request that failed mid-stream has no usable response_obj usage, but the
534 # streaming handler may have recovered the usage from the chunks already
535 # delivered. Honor that override so the partial usage lands in spend tracking.
536 _combined_usage: Final = kwargs.get("combined_usage_object")
537 if not usage and isinstance(_combined_usage, litellm.Usage): 537 ↛ 538line 537 didn't jump to line 538 because the condition on line 537 was never true
538 usage = _combined_usage.model_dump()
540 id = get_spend_logs_id(call_type or "acompletion", response_obj_dict, kwargs)
541 standard_logging_payload: Final = cast(StandardLoggingPayload | None, kwargs.get("standard_logging_object", None))
543 end_user_id = get_end_user_id_for_cost_tracking(litellm_params)
545 api_key = metadata.get("user_api_key", "")
547 standard_logging_prompt_tokens: int = 0
548 standard_logging_completion_tokens: int = 0
549 standard_logging_total_tokens: int = 0
550 if standard_logging_payload is not None:
551 standard_logging_prompt_tokens = standard_logging_payload.get("prompt_tokens", 0)
552 standard_logging_completion_tokens = standard_logging_payload.get("completion_tokens", 0)
553 standard_logging_total_tokens = standard_logging_payload.get("total_tokens", 0)
554 _trusted_hash = metadata.get("user_api_key_hash")
555 _key_alias = metadata.get("user_api_key_alias")
556 _key_already_redacted = (
557 isinstance(_trusted_hash, str)
558 and _is_non_secret_key_value(_trusted_hash, key_alias=_key_alias)
559 and _trusted_hash == api_key
560 )
561 api_key = _redact_logged_api_key(api_key, already_redacted=_key_already_redacted, key_alias=_key_alias) or ""
563 if (
564 standard_logging_payload is not None
565 ): # [TODO] migrate completely to sl payload. currently missing pass-through endpoint data
566 api_key = (
567 api_key
568 or _redact_logged_api_key(
569 standard_logging_payload["metadata"].get("user_api_key_hash"),
570 already_redacted=True,
571 key_alias=standard_logging_payload["metadata"].get("user_api_key_alias"),
572 )
573 or ""
574 )
575 end_user_id = end_user_id or standard_logging_payload["metadata"].get("user_api_key_end_user_id")
576 request_tags = safe_dumps(metadata.get("tags", [])) if isinstance(metadata.get("tags", []), list) else "[]"
577 if (
578 standard_logging_payload is not None and standard_logging_payload.get("request_tags") is not None
579 ): # use 'tags' from standard logging payload instead
580 request_tags = safe_dumps(standard_logging_payload["request_tags"])
582 _model_id: Final = metadata.get("model_info", {}).get("id", "") or _sl_attribution_fallback(
583 standard_logging_payload, "model_id"
584 )
585 _model_group: Final = metadata.get("model_group", "") or _sl_attribution_fallback(
586 standard_logging_payload, "model_group"
587 )
588 _api_base: Final = litellm_params.get("api_base", "") or _sl_attribution_fallback(
589 standard_logging_payload, "api_base"
590 )
592 # Extract overhead from hidden_params if available
593 litellm_overhead_time_ms = None
594 if standard_logging_payload is not None:
595 hidden_params: Final = standard_logging_payload.get("hidden_params", {})
596 litellm_overhead_time_ms = hidden_params.get("litellm_overhead_time_ms")
598 logged_provider: Final = (
599 kwargs.get("custom_llm_provider")
600 or _sl_attribution_fallback(standard_logging_payload, "custom_llm_provider")
601 or None
602 )
603 custom_llm_provider: Final = logged_provider or _model_group_provider(_model_group, llm_router)
604 requested_model: Final = cast(object, kwargs.get("model"))
605 raw_model: Final = requested_model if isinstance(requested_model, str) else ""
606 model_is_malformed: Final = requested_model is not None and not isinstance(requested_model, str)
607 logged_model: Final = standard_logging_payload.get("model") if standard_logging_payload is not None else None
608 resolved_model: Final = (logged_model if isinstance(logged_model, str) else None) or reconstruct_model_name(
609 raw_model, logged_provider, metadata or {}
610 )
611 failed_with_prompt_shaped_model: Final = (
612 _get_status_for_spend_log(metadata=metadata) == "failure"
613 and not _model_id
614 and not _looks_like_model_name(resolved_model)
615 and not _is_configured_model_group(_model_group, llm_router)
616 )
617 model_name: Final = (
618 UNKNOWN_MODEL_SPEND_LOG_MODEL
619 if rejected_as_unknown_model or failed_with_prompt_shaped_model or model_is_malformed
620 else resolved_model
621 )
622 model_is_placeholdered: Final = model_name == UNKNOWN_MODEL_SPEND_LOG_MODEL
623 persisted_model_group: Final = (
624 ""
625 if model_is_placeholdered and _model_group == raw_model and not _looks_like_model_name(raw_model)
626 else _model_group
627 )
628 persisted_metadata: Final = (
629 {
630 **metadata,
631 "error_information": _scrub_raw_model_from_error_information(metadata.get("error_information"), raw_model),
632 }
633 if model_is_placeholdered
634 else metadata
635 )
636 litellm_call_id: Final = cast(
637 str | None,
638 kwargs.get("litellm_call_id") or litellm_params.get("litellm_call_id"),
639 )
641 # clean up litellm metadata
642 clean_metadata = _get_spend_logs_metadata(
643 persisted_metadata,
644 applied_guardrails=(
645 standard_logging_payload["metadata"].get("applied_guardrails", None)
646 if standard_logging_payload is not None
647 else None
648 ),
649 batch_models=(
650 standard_logging_payload.get("hidden_params", {}).get("batch_models", None)
651 if standard_logging_payload is not None
652 else None
653 ),
654 batch_successful_requests=(
655 standard_logging_payload.get("hidden_params", {}).get("batch_successful_requests", None)
656 if standard_logging_payload is not None
657 else None
658 ),
659 batch_failed_requests=(
660 standard_logging_payload.get("hidden_params", {}).get("batch_failed_requests", None)
661 if standard_logging_payload is not None
662 else None
663 ),
664 mcp_tool_call_metadata=(
665 standard_logging_payload["metadata"].get("mcp_tool_call_metadata", None)
666 if standard_logging_payload is not None
667 else None
668 ),
669 vector_store_request_metadata=(
670 standard_logging_payload["metadata"].get("vector_store_request_metadata", None)
671 if standard_logging_payload is not None
672 else None
673 ),
674 usage_object=(
675 standard_logging_payload["metadata"].get("usage_object", None)
676 if standard_logging_payload is not None
677 else None
678 ),
679 model_map_information=(
680 standard_logging_payload["model_map_information"] if standard_logging_payload is not None else None
681 ),
682 guardrail_information=(
683 standard_logging_payload.get("guardrail_information", None)
684 if standard_logging_payload is not None
685 else (metadata.get("standard_logging_guardrail_information", None) if metadata is not None else None)
686 ),
687 cold_storage_object_key=(
688 standard_logging_payload["metadata"].get("cold_storage_object_key", None)
689 if standard_logging_payload is not None
690 else None
691 ),
692 litellm_overhead_time_ms=litellm_overhead_time_ms,
693 cost_breakdown=(
694 standard_logging_payload.get("cost_breakdown", None) if standard_logging_payload is not None else None
695 ),
696 autorouter_savings=(
697 standard_logging_payload.get("autorouter_savings", None) if standard_logging_payload is not None else None
698 ),
699 autorouter_savings_estimate=(
700 standard_logging_payload.get("autorouter_savings_estimate")
701 if standard_logging_payload is not None
702 else None
703 ),
704 autorouter_baseline_observation=(
705 standard_logging_payload.get("autorouter_baseline_observation")
706 if standard_logging_payload is not None
707 else None
708 ),
709 litellm_call_id=litellm_call_id,
710 router_metadata=_get_router_metadata_for_spend_log(
711 metadata=metadata,
712 requested_model=persisted_model_group,
713 selected_model=model_name,
714 selected_provider=custom_llm_provider,
715 router_correlation_id=litellm_call_id,
716 ),
717 azure_spillover=azure_spillover(
718 response_headers=kwargs.get("response_headers")
719 if isinstance(kwargs.get("response_headers"), Mapping)
720 else None,
721 additional_headers=standard_logging_payload["hidden_params"].get("additional_headers")
722 if standard_logging_payload is not None
723 and isinstance(standard_logging_payload.get("hidden_params"), Mapping)
724 else None,
725 ),
726 )
728 special_usage_fields: Final = ["completion_tokens", "prompt_tokens", "total_tokens"]
729 additional_usage_values = {}
730 for k, v in usage.items(): 730 ↛ 731line 730 didn't jump to line 731 because the loop on line 730 never started
731 if k not in special_usage_fields:
732 if isinstance(v, BaseModel):
733 v = v.model_dump()
734 additional_usage_values.update({k: v})
735 prompt_tokens_details = additional_usage_values.get("prompt_tokens_details")
736 if not isinstance(prompt_tokens_details, dict): 736 ↛ 740line 736 didn't jump to line 740 because the condition on line 736 was always true
737 usage_object: Final = clean_metadata.get("usage_object")
738 if isinstance(usage_object, dict):
739 prompt_tokens_details = usage_object.get("prompt_tokens_details")
740 if isinstance(prompt_tokens_details, dict): 740 ↛ 741line 740 didn't jump to line 741 because the condition on line 740 was never true
741 if "cache_read_input_tokens" not in additional_usage_values:
742 cached_tokens: Final = prompt_tokens_details.get("cached_tokens")
743 if isinstance(cached_tokens, int) and cached_tokens > 0:
744 additional_usage_values["cache_read_input_tokens"] = cached_tokens
745 if "cache_creation_input_tokens" not in additional_usage_values:
746 cache_write_tokens: Final = prompt_tokens_details.get("cache_write_tokens") or prompt_tokens_details.get(
747 "cache_creation_tokens"
748 )
749 if isinstance(cache_write_tokens, int) and cache_write_tokens > 0:
750 additional_usage_values["cache_creation_input_tokens"] = cache_write_tokens
751 clean_metadata["additional_usage_values"] = additional_usage_values
753 if litellm.cache is None:
754 cache_key = "Cache OFF"
755 elif litellm_params.get("preset_cache_key") is not None: 755 ↛ 756line 755 didn't jump to line 756 because the condition on line 755 was never true
756 cache_key = litellm_params["preset_cache_key"]
757 else:
758 cache_key = litellm.cache.get_cache_key(**kwargs)
759 if cache_hit is True: 759 ↛ 760line 759 didn't jump to line 760 because the condition on line 759 was never true
760 import time
762 id = f"{id}_cache_hit{time.time()}" # SpendLogs does not allow duplicate request_id
764 mcp_namespaced_tool_name = None
765 mcp_tool_call_metadata: Final[StandardLoggingMCPToolCall | None] = clean_metadata.get("mcp_tool_call_metadata")
766 if mcp_tool_call_metadata is not None: 766 ↛ 767line 766 didn't jump to line 767 because the condition on line 766 was never true
767 mcp_namespaced_tool_name = mcp_tool_call_metadata.get("namespaced_tool_name", None)
769 # Extract agent_id for A2A requests (set directly on model_call_details)
770 agent_id: Final[str | None] = kwargs.get("agent_id") or metadata.get("agent_id")
772 try:
773 payload: Final[SpendLogsPayload] = SpendLogsPayload(
774 request_id=str(id),
775 call_type=call_type or "",
776 api_key=str(api_key),
777 cache_hit=str(cache_hit),
778 startTime=_ensure_datetime_utc(start_time),
779 endTime=_ensure_datetime_utc(end_time),
780 completionStartTime=_ensure_datetime_utc(completion_start_time),
781 model=model_name,
782 user=metadata.get("user_api_key_user_id", "") or "",
783 team_id=metadata.get("user_api_key_team_id", "") or "",
784 organization_id=metadata.get("user_api_key_org_id") or "",
785 metadata=safe_dumps(clean_metadata),
786 cache_key=cache_key,
787 spend=kwargs.get("response_cost", 0),
788 total_tokens=usage.get("total_tokens", standard_logging_total_tokens),
789 prompt_tokens=usage.get("prompt_tokens", standard_logging_prompt_tokens),
790 completion_tokens=usage.get("completion_tokens", standard_logging_completion_tokens),
791 request_tags=request_tags,
792 end_user=end_user_id or "",
793 api_base=_api_base,
794 model_group=persisted_model_group,
795 model_id=_model_id,
796 mcp_namespaced_tool_name=mcp_namespaced_tool_name,
797 agent_id=agent_id,
798 requester_ip_address=clean_metadata.get("requester_ip_address", None),
799 custom_llm_provider=custom_llm_provider or "",
800 messages=_get_messages_for_spend_logs_payload(
801 standard_logging_payload=standard_logging_payload, metadata=metadata
802 ),
803 response=_get_response_for_spend_logs_payload(payload=standard_logging_payload, kwargs=kwargs),
804 proxy_server_request=_get_proxy_server_request_for_spend_logs_payload(
805 metadata=metadata,
806 litellm_params=(
807 _placeholder_stored_request_body(litellm_params, persisted_model_group, raw_model)
808 if model_is_placeholdered
809 else litellm_params
810 ),
811 kwargs=kwargs,
812 ),
813 session_id=_get_session_id_for_spend_log(
814 kwargs=kwargs,
815 metadata=metadata,
816 standard_logging_payload=standard_logging_payload,
817 omit_when_missing=_omits_session_id_when_missing(metadata),
818 batch_trace_session_id=_get_batch_trace_session_id(call_type=call_type, request_id=id),
819 ),
820 request_duration_ms=_get_request_duration_ms(start_time, end_time),
821 status=_get_status_for_spend_log(
822 metadata=metadata,
823 ),
824 litellm_call_id=litellm_call_id,
825 )
827 verbose_proxy_logger.debug(
828 "SpendTable: created payload - request_id: %s, model: %s, spend: %s",
829 payload.get("request_id"),
830 payload.get("model"),
831 payload.get("spend"),
832 )
834 # Explicitly clear large intermediate objects to reduce memory pressure
835 del response_obj_dict, usage, clean_metadata, additional_usage_values
837 return payload
838 except Exception as e:
839 spend_log_error("Error creating spendlogs object - %s", str(e), exc=e)
840 raise e
843def _omits_session_id_when_missing(metadata: Mapping[str, object] | None) -> bool:
844 """The pre-call stamp pins `omit` on for the requests that carry it, so a config reload between pre-call and spend
845 logging cannot fabricate a session. `apply_missing_session_id_policy` drops any client-supplied copy of the key
846 from both metadata buckets before stamping, which the merge of `litellm_metadata` into `metadata` makes
847 necessary, so a caller cannot forge it. Requests that never reach the pre-call helper, router-model
848 passthrough among them, carry no stamp, so they fall back to the configured policy and `omit` still covers their
849 spend logs."""
850 if metadata is not None and metadata.get(SESSION_ID_OMITTED_METADATA_KEY): 850 ↛ 851line 850 didn't jump to line 851 because the condition on line 850 was never true
851 return True
853 from litellm.proxy.proxy_server import general_settings
855 return general_settings.get("missing_session_id") == "omit"
858_BATCH_TRACE_CALL_TYPES: Final = frozenset(
859 {
860 CallTypes.create_batch.value,
861 CallTypes.acreate_batch.value,
862 CallTypes.retrieve_batch.value,
863 CallTypes.aretrieve_batch.value,
864 }
865)
868def _get_batch_trace_session_id(call_type: str | None, request_id: str | None) -> str | None:
869 """A batch's create row and its poller-written cost row both derive their request id
870 from the same batch id (the cost row appends BATCH_COST_REQUEST_ID_SUFFIX), so using
871 that id as the session groups the batch lifecycle into one trace on the logs UI. The
872 poller builds its own logging context, so per-request trace ids can never link them."""
873 if call_type not in _BATCH_TRACE_CALL_TYPES or not request_id:
874 return None
875 return request_id.removesuffix(BATCH_COST_REQUEST_ID_SUFFIX)
878def _get_session_id_for_spend_log(
879 kwargs: Mapping[str, object],
880 metadata: Mapping[str, object] | None,
881 standard_logging_payload: StandardLoggingPayload | None,
882 omit_when_missing: bool,
883 batch_trace_session_id: str | None = None,
884) -> str | None:
885 """Under `omit` only `metadata.session_id`, the key Langfuse reads, counts as a session; `litellm_session_id` may
886 be a copied trace id. Batch call types carry a deterministic session derived from the batch id, which outranks
887 the per-request trace ids because those differ between the create call and the cost poller's row."""
888 if omit_when_missing: 888 ↛ 889line 888 didn't jump to line 889 because the condition on line 888 was never true
889 session_id: Final = metadata.get("session_id") if metadata else None
890 return str(session_id) if session_id else None
892 from litellm._uuid import uuid
894 if batch_trace_session_id is not None:
895 return batch_trace_session_id
896 if standard_logging_payload is not None and standard_logging_payload.get("trace_id") is not None:
897 return str(standard_logging_payload.get("trace_id"))
898 if kwargs.get("litellm_trace_id") is not None: 898 ↛ 899line 898 didn't jump to line 899 because the condition on line 898 was never true
899 return str(kwargs.get("litellm_trace_id"))
900 return str(uuid.uuid4())
903def _get_request_duration_ms(start_time: datetime, end_time: datetime) -> int | None:
904 """Compute request duration in milliseconds from start and end times."""
905 try:
906 return int((end_time - start_time).total_seconds() * 1000)
907 except Exception:
908 return None
911def _ensure_datetime_utc(timestamp: datetime) -> datetime:
912 """Helper to ensure datetime is in UTC"""
913 timestamp = timestamp.astimezone(timezone.utc)
914 return timestamp
917async def _query_raw_rows(
918 prisma_client: PrismaClient,
919 sql_query: str,
920 *args: object,
921) -> Sequence[Mapping[str, object]] | None:
922 return await prisma_client.db.query_raw(sql_query, *args)
925async def get_spend_by_team(
926 start_date: dt,
927 end_date: dt,
928 team_id: str | None,
929 prisma_client: PrismaClient,
930):
931 sql_query: Final = """
932 WITH SpendByModelApiKey AS (
933 SELECT
934 date_trunc('day', sl."startTime") AS group_by_day,
935 COALESCE(tt.team_alias, 'Unassigned Team') AS team_name,
936 sl.model,
937 sl.api_key,
938 SUM(sl.spend) AS model_api_spend,
939 SUM(sl.total_tokens) AS model_api_tokens
940 FROM
941 "LiteLLM_SpendLogs" sl
942 LEFT JOIN
943 "LiteLLM_TeamTable" tt
944 ON
945 sl.team_id = tt.team_id
946 WHERE
947 sl."startTime" >= ($1::timestamptz AT TIME ZONE 'UTC')
948 AND sl."startTime" < (($2::timestamptz + INTERVAL '1 day') AT TIME ZONE 'UTC')
949 AND ($3::text IS NULL OR sl.team_id = $3)
950 GROUP BY
951 date_trunc('day', sl."startTime"),
952 tt.team_alias,
953 sl.model,
954 sl.api_key
955 )
956 SELECT
957 group_by_day,
958 jsonb_agg(jsonb_build_object(
959 'team_name', team_name,
960 'total_spend', total_spend,
961 'metadata', metadata
962 )) AS teams
963 FROM (
964 SELECT
965 group_by_day,
966 team_name,
967 SUM(model_api_spend) AS total_spend,
968 jsonb_agg(jsonb_build_object(
969 'model', model,
970 'api_key', api_key,
971 'spend', model_api_spend,
972 'total_tokens', model_api_tokens
973 )) AS metadata
974 FROM
975 SpendByModelApiKey
976 GROUP BY
977 group_by_day,
978 team_name
979 ) AS aggregated
980 GROUP BY
981 group_by_day
982 ORDER BY
983 group_by_day;
984 """
986 db_response: Final = await _query_raw_rows(prisma_client, sql_query, start_date, end_date, team_id)
987 if db_response is None:
988 return []
990 return db_response
993async def get_spend_by_team_and_customer(
994 start_date: dt,
995 end_date: dt,
996 team_id: str,
997 customer_id: str,
998 prisma_client: PrismaClient,
999):
1000 sql_query: Final = """
1001 WITH SpendByModelApiKey AS (
1002 SELECT
1003 date_trunc('day', sl."startTime") AS group_by_day,
1004 COALESCE(tt.team_alias, 'Unassigned Team') AS team_name,
1005 sl.end_user AS customer,
1006 sl.model,
1007 sl.api_key,
1008 SUM(sl.spend) AS model_api_spend,
1009 SUM(sl.total_tokens) AS model_api_tokens
1010 FROM
1011 "LiteLLM_SpendLogs" sl
1012 LEFT JOIN
1013 "LiteLLM_TeamTable" tt
1014 ON
1015 sl.team_id = tt.team_id
1016 WHERE
1017 sl."startTime" >= ($1::timestamptz AT TIME ZONE 'UTC')
1018 AND sl."startTime" < (($2::timestamptz + INTERVAL '1 day') AT TIME ZONE 'UTC')
1019 AND sl.team_id = $3
1020 AND sl.end_user = $4
1021 GROUP BY
1022 date_trunc('day', sl."startTime"),
1023 tt.team_alias,
1024 sl.end_user,
1025 sl.model,
1026 sl.api_key
1027 )
1028 SELECT
1029 group_by_day,
1030 jsonb_agg(jsonb_build_object(
1031 'team_name', team_name,
1032 'customer', customer,
1033 'total_spend', total_spend,
1034 'metadata', metadata
1035 )) AS teams_customers
1036 FROM (
1037 SELECT
1038 group_by_day,
1039 team_name,
1040 customer,
1041 SUM(model_api_spend) AS total_spend,
1042 jsonb_agg(jsonb_build_object(
1043 'model', model,
1044 'api_key', api_key,
1045 'spend', model_api_spend,
1046 'total_tokens', model_api_tokens
1047 )) AS metadata
1048 FROM
1049 SpendByModelApiKey
1050 GROUP BY
1051 group_by_day,
1052 team_name,
1053 customer
1054 ) AS aggregated
1055 GROUP BY
1056 group_by_day
1057 ORDER BY
1058 group_by_day;
1059 """
1061 db_response: Final = await _query_raw_rows(prisma_client, sql_query, start_date, end_date, team_id, customer_id)
1062 if db_response is None:
1063 return []
1065 return db_response
1068def _get_messages_for_spend_logs_payload(
1069 standard_logging_payload: StandardLoggingPayload | None,
1070 metadata: dict | None = None,
1071) -> str:
1072 if should_store_prompts_and_responses_in_spend_logs(): 1072 ↛ 1073line 1072 didn't jump to line 1073 because the condition on line 1072 was never true
1073 if standard_logging_payload is not None:
1074 call_type: Final = standard_logging_payload.get("call_type", "")
1075 if call_type == "_arealtime":
1076 messages: Final = standard_logging_payload.get("messages")
1077 if messages is not None:
1078 try:
1079 return safe_dumps(messages)
1080 except Exception:
1081 return "{}"
1082 return "{}"
1085_SENSITIVE_REQUEST_BODY_KEYS: Final = frozenset({"secret_fields"})
1088def _sanitize_request_body_for_spend_logs_payload(
1089 request_body: Mapping[str, object],
1090 visited: set | None = None,
1091 max_string_length_prompt_in_db: int | None = None,
1092) -> dict:
1093 """
1094 Recursively sanitize request body to prevent logging large base64 strings or other large values.
1095 Truncates strings longer than MAX_STRING_LENGTH_PROMPT_IN_DB characters and handles nested dictionaries.
1097 Also strips keys listed in _SENSITIVE_REQUEST_BODY_KEYS (e.g. secret_fields
1098 which contains raw HTTP headers including Authorization tokens).
1099 """
1100 from litellm.constants import (
1101 LITELLM_TRUNCATED_PAYLOAD_FIELD,
1102 LITELLM_TRUNCATION_DB_SAFEGUARD_NOTE,
1103 )
1105 if visited is None: 1105 ↛ 1107line 1105 didn't jump to line 1107 because the condition on line 1105 was always true
1106 visited = set()
1107 if max_string_length_prompt_in_db is None: 1107 ↛ 1111line 1107 didn't jump to line 1111 because the condition on line 1107 was always true
1108 max_string_length_prompt_in_db = _get_max_string_length_prompt_in_db()
1110 # Get the object's memory address to track visited objects
1111 obj_id: Final = id(request_body)
1112 if obj_id in visited: 1112 ↛ 1113line 1112 didn't jump to line 1113 because the condition on line 1112 was never true
1113 return {}
1114 visited.add(obj_id)
1116 def _sanitize_value(value: object) -> object:
1117 if isinstance(value, Mapping): 1117 ↛ 1118line 1117 didn't jump to line 1118 because the condition on line 1117 was never true
1118 return _sanitize_request_body_for_spend_logs_payload(value, visited, max_string_length_prompt_in_db)
1119 elif isinstance(value, list): 1119 ↛ 1120line 1119 didn't jump to line 1120 because the condition on line 1119 was never true
1120 return [_sanitize_value(item) for item in value]
1121 elif isinstance(value, str):
1122 if len(value) > max_string_length_prompt_in_db:
1123 # Keep 35% from beginning and 65% from end (end is usually more important)
1124 # This split ensures we keep more context from the end of conversations
1125 start_ratio: Final = 0.35
1126 end_ratio: Final = 0.65
1128 # Calculate character distribution
1129 start_chars: Final = int(max_string_length_prompt_in_db * start_ratio)
1130 end_chars = int(max_string_length_prompt_in_db * end_ratio)
1132 # Ensure we don't exceed the total limit
1133 total_keep: Final = start_chars + end_chars
1134 if total_keep > max_string_length_prompt_in_db: 1134 ↛ 1135line 1134 didn't jump to line 1135 because the condition on line 1134 was never true
1135 end_chars = max_string_length_prompt_in_db - start_chars
1137 # If the string length is less than what we want to keep, just truncate normally
1138 if len(value) <= max_string_length_prompt_in_db: 1138 ↛ 1139line 1138 didn't jump to line 1139 because the condition on line 1138 was never true
1139 return value
1141 # Calculate how many characters are being skipped
1142 skipped_chars: Final = len(value) - total_keep
1144 # Build the truncated string: beginning + truncation marker + end
1145 truncated_value: Final = (
1146 f"{value[:start_chars]}"
1147 f"... ({LITELLM_TRUNCATED_PAYLOAD_FIELD} skipped {skipped_chars} chars. "
1148 f"{LITELLM_TRUNCATION_DB_SAFEGUARD_NOTE}) ..."
1149 f"{value[-end_chars:]}"
1150 )
1151 return truncated_value
1152 return value
1153 return value
1155 return {k: _sanitize_value(v) for k, v in request_body.items() if k not in _SENSITIVE_REQUEST_BODY_KEYS}
1158# Quoted-key form: ``"input"`` / ``'messages'`` / ``"prompt"`` followed by
1159# ``:``. Covers JSON bodies and Python dict-reprs in provider error strings.
1160# ``prompt`` is included for ``/v1/completions``-style payloads where the user
1161# input lives under a top-level ``prompt`` key rather than ``messages``.
1162_ERROR_MESSAGE_PROMPT_LEAK_KEYS: Final = ("input", "messages", "prompt")
1165# Assignment-style keys: Pydantic v2 validation errors render the offending
1166# value as ``input_value=<repr>`` inside ``[type=..., input_value=...,
1167# input_type=...]``. The same prompt body that would appear under an
1168# ``"input"`` JSON key is echoed here as a Python repr, so we redact it
1169# under the same store_prompts_in_spend_logs gate.
1170_ERROR_MESSAGE_ASSIGN_LEAK_KEYS: Final = ("input_value",)
1173_SENSITIVE_KEY_START_PATTERN: Final = re.compile(
1174 r"(?:"
1175 r"['\"](?:" + "|".join(_ERROR_MESSAGE_PROMPT_LEAK_KEYS) + r")['\"]\s*:\s*"
1176 r"|"
1177 r"\b(?:" + "|".join(_ERROR_MESSAGE_ASSIGN_LEAK_KEYS) + r")\s*=\s*"
1178 r")"
1179)
1182def _scan_quoted_string_end(text: str, start: int, quote: str) -> int:
1183 """
1184 Given ``text[start] == quote`` (``'`` or ``"``), return the index just
1185 past the matching close quote, honoring backslash escapes. Returns
1186 ``-1`` if unterminated.
1187 """
1188 n: Final = len(text)
1189 i = start + 1
1190 while i < n:
1191 c = text[i]
1192 if c == "\\":
1193 i += 2
1194 continue
1195 if c == quote:
1196 return i + 1
1197 i += 1
1198 return -1
1201def _scan_balanced_value_end(text: str, start: int) -> int:
1202 """
1203 Given ``text[start]`` is ``[``, ``{``, ``'`` or ``"``, return the index
1204 just past the matching close, accounting for nested brackets and
1205 quoted strings (with escape sequences). Returns ``-1`` if the
1206 structure is unterminated.
1208 Implemented iteratively (no self-recursion): the bracket scanner
1209 inlines a quote-skip helper rather than re-entering itself, since
1210 JSON-style values cannot contain another bracket *as a first char*
1211 inside a quoted string — only the quote-skip case can occur.
1212 """
1213 n: Final = len(text)
1214 if start >= n:
1215 return -1
1216 first: Final = text[start]
1217 if first in ("'", '"'):
1218 return _scan_quoted_string_end(text, start, first)
1219 if first == "[":
1220 close = "]"
1221 elif first == "{":
1222 close = "}"
1223 else:
1224 return -1
1225 depth = 0
1226 i = start
1227 while i < n:
1228 c = text[i]
1229 if c in ("'", '"'):
1230 end = _scan_quoted_string_end(text, i, c)
1231 if end == -1:
1232 return -1
1233 i = end
1234 continue
1235 if c == first:
1236 depth += 1
1237 elif c == close:
1238 depth -= 1
1239 if depth == 0:
1240 return i + 1
1241 i += 1
1242 return -1
1245def _redact_prompt_leaks_in_error_string(text: str) -> str:
1246 """
1247 Strip echoed request input from provider error strings.
1249 Provider validation errors (e.g. OpenAI ``RateLimitError`` carrying 178
1250 pydantic validation errors, each with its own ``'input': [...]`` field)
1251 embed the full request body in their message. When prompts must not be
1252 stored in spend logs, that echo is a back-door leak.
1254 Two leak shapes are handled:
1256 - Quoted-key form — ``"<key>": <value>`` where ``key`` is ``input``,
1257 ``messages`` or ``prompt`` (covers JSON bodies, Python dict-reprs,
1258 and ``/v1/completions`` payloads).
1259 - Assignment form — ``input_value=<value>`` from Pydantic v2 validation
1260 errors, which render the offending value as a Python repr inside
1261 ``[type=..., input_value=..., input_type=...]``.
1263 The value scan understands nested ``[]`` / ``{}`` and quoted strings,
1264 so multi-modal payloads (``'messages': [{'content': [{...}]}]``) and
1265 user text containing brackets (``"secret[123"``) are handled correctly.
1266 """
1267 if not text:
1268 return text
1269 redaction: Final = f'"{REDACTED_BY_LITELM_STRING}"'
1270 out: Final[list[str]] = []
1271 n: Final = len(text)
1272 pos = 0
1273 while pos < n: 1273 ↛ 1297line 1273 didn't jump to line 1297 because the condition on line 1273 was always true
1274 m = _SENSITIVE_KEY_START_PATTERN.search(text, pos)
1275 if not m:
1276 out.append(text[pos:])
1277 break
1278 out.append(text[pos : m.end()])
1279 v_start = m.end()
1280 if v_start >= n: 1280 ↛ 1281line 1280 didn't jump to line 1281 because the condition on line 1280 was never true
1281 break
1282 first = text[v_start]
1283 if first in ("[", "{", "'", '"'): 1283 ↛ 1284line 1283 didn't jump to line 1284 because the condition on line 1283 was never true
1284 v_end = _scan_balanced_value_end(text, v_start)
1285 if v_end == -1:
1286 # Unterminated value — redact through the rest of the string
1287 # so a malformed leak can't slip past.
1288 out.append(redaction)
1289 pos = n
1290 break
1291 out.append(redaction)
1292 pos = v_end
1293 else:
1294 # Unquoted scalar (number, null, bare identifier) — not a leak
1295 # carrier, leave intact and resume after the key match.
1296 pos = v_start
1297 return "".join(out)
1300def _sanitize_guardrail_information_for_spend_logs(
1301 guardrail_information: list[StandardLoggingGuardrailInformation] | None,
1302) -> list[StandardLoggingGuardrailInformation] | None:
1303 """
1304 When ``store_prompts_in_spend_logs`` is False, redact prompt-carrying fields
1305 (``guardrail_request``, ``guardrail_response``, ``match_details``,
1306 ``classification``) before they land in ``LiteLLM_SpendLogs.metadata``.
1308 Guardrail hooks may echo the LLM request payload back into
1309 ``guardrail_response``, and two first-party hooks
1310 (``block_code_execution``, ``litellm_content_filter``) inline user-prompt
1311 substrings into ``match_details`` / ``classification`` too, so the flag
1312 must cover all four fields. Every other typed field on the entry (name,
1313 provider, mode, status, timings, action, violation_categories, risk_score,
1314 masked_entity_count, ...) is preserved so guardrail dashboards keep
1315 working.
1317 ``guardrail_information`` is typed ``Optional[List[...]]`` but at least
1318 one writer (``xecguard``) assigns a bare dict, so normalize to a list
1319 here to match OTEL's defensive read pattern; otherwise iteration would
1320 yield the dict's keys and crash the whole spend-log write.
1321 """
1322 if guardrail_information is None or should_store_prompts_and_responses_in_spend_logs(): 1322 ↛ 1324line 1322 didn't jump to line 1324 because the condition on line 1322 was always true
1323 return guardrail_information
1324 entries: Final = [guardrail_information] if isinstance(guardrail_information, dict) else guardrail_information
1325 return [_redact_prompt_fields_in_guardrail_entry(entry) for entry in entries if isinstance(entry, dict)]
1328_NUMERIC_COMPRESSION_STAT_KEYS: Final = (
1329 "tokens_before",
1330 "tokens_after",
1331 "tokens_saved",
1332 "compression_ratio",
1333)
1336def _numeric_compression_stats_from_guardrail_response(
1337 guardrail_response: object,
1338) -> dict[str, int | float] | None:
1339 if not isinstance(guardrail_response, dict):
1340 return None
1341 stats: Final = {
1342 key: value
1343 for key, value in guardrail_response.items()
1344 if key in _NUMERIC_COMPRESSION_STAT_KEYS and isinstance(value, (int, float)) and not isinstance(value, bool)
1345 }
1346 return stats or None
1349def _redact_prompt_fields_in_guardrail_entry(
1350 entry: StandardLoggingGuardrailInformation,
1351) -> StandardLoggingGuardrailInformation:
1352 """
1353 Replace prompt-carrying fields with the redaction marker. Purely numeric
1354 compression stats inside ``guardrail_response`` (e.g. Headroom's
1355 ``tokens_saved``) cannot carry prompt content, so they are preserved as a
1356 stats-only dict; spend aggregation reads them via
1357 ``extract_compression_saved_tokens``.
1358 """
1359 preserved_stats: Final = _numeric_compression_stats_from_guardrail_response(entry.get("guardrail_response"))
1360 redacted: Final[StandardLoggingGuardrailInformation] = {
1361 **entry,
1362 **{key: REDACTED_BY_LITELM_STRING for key in PROMPT_CARRYING_GUARDRAIL_FIELDS if key in entry},
1363 }
1364 if preserved_stats is None:
1365 return redacted
1366 return {**redacted, "guardrail_response": preserved_stats}
1369def _sanitize_error_information_for_spend_logs(
1370 error_information: StandardLoggingPayloadErrorInformation | None,
1371 original_exception: BaseException | None = None,
1372) -> StandardLoggingPayloadErrorInformation | None:
1373 """
1374 Sanitize ``error_information`` before it lands in ``LiteLLM_SpendLogs.metadata``.
1376 Provider errors are stored verbatim via ``str(original_exception)``; those
1377 strings can echo the full request body, producing multi-megabyte spend-log
1378 rows.
1380 - Always: cap ``error_message`` and ``traceback`` with the existing
1381 ``MAX_STRING_LENGTH_PROMPT_IN_DB`` DB-storage safeguard.
1382 - When ``store_prompts_in_spend_logs`` is False: additionally redact
1383 ``'input'`` / ``'messages'`` / ``'prompt'`` values *and* Pydantic v2
1384 ``input_value=...`` assignments inside both ``error_message`` and
1385 ``traceback`` so prompts cannot leak through either field.
1387 Scoped to the spend-log path — OTEL/Datadog/etc. callbacks still receive
1388 the untruncated error per ``LITELLM_TRUNCATION_DB_SAFEGUARD_NOTE``.
1389 """
1390 if error_information is None: 1390 ↛ 1391line 1390 didn't jump to line 1391 because the condition on line 1390 was never true
1391 return None
1393 persisted: Final = (
1394 {**error_information, "error_message": original_exception.spend_log_error_message}
1395 if isinstance(original_exception, ProxyModelNotFoundError)
1396 else error_information
1397 )
1398 sanitized = cast(dict, {**persisted})
1400 if not should_store_prompts_and_responses_in_spend_logs(): 1400 ↛ 1406line 1400 didn't jump to line 1406 because the condition on line 1400 was always true
1401 for field in ("error_message", "traceback"):
1402 value = sanitized.get(field)
1403 if isinstance(value, str): 1403 ↛ 1401line 1403 didn't jump to line 1401 because the condition on line 1403 was always true
1404 sanitized[field] = _redact_prompt_leaks_in_error_string(value)
1406 sanitized = _sanitize_request_body_for_spend_logs_payload(sanitized)
1407 return cast(StandardLoggingPayloadErrorInformation, sanitized)
1410def _convert_to_json_serializable_dict(obj: object, visited: set[int] | None = None, max_depth: int = 20) -> object:
1411 """
1412 Convert object to JSON-serializable dict, handling Pydantic models safely.
1414 This avoids pickle-based deepcopy which fails on Pydantic v2 models
1415 containing _thread.RLock objects.
1417 Args:
1418 obj: Object to convert (dict, list, Pydantic model, or primitive)
1419 visited: Set of object IDs to track circular references
1420 max_depth: Maximum recursion depth to prevent infinite recursion
1422 Returns:
1423 JSON-serializable version of the object
1424 """
1425 if max_depth <= 0:
1426 # Return a placeholder if max depth is exceeded
1427 return "<max_depth_exceeded>"
1429 if visited is None:
1430 visited = set()
1432 # Get the object's memory address to track visited objects
1433 obj_id: Final = id(obj)
1434 if obj_id in visited:
1435 # Circular reference detected, return placeholder
1436 return "<circular_reference>"
1438 # Only track mutable objects (dict, list, objects with __dict__)
1439 if isinstance(obj, (dict, list)) or hasattr(obj, "__dict__"):
1440 visited.add(obj_id)
1442 try:
1443 if isinstance(obj, BaseModel):
1444 # Use Pydantic's model_dump() instead of pickle
1445 result: Final = obj.model_dump()
1446 # Recursively process the dumped dict
1447 return _convert_to_json_serializable_dict(result, visited, max_depth - 1)
1448 elif isinstance(obj, dict):
1449 return {k: _convert_to_json_serializable_dict(v, visited, max_depth - 1) for k, v in obj.items()}
1450 elif isinstance(obj, list):
1451 return [_convert_to_json_serializable_dict(item, visited, max_depth - 1) for item in obj]
1452 elif hasattr(obj, "__dict__"):
1453 # Handle objects with __dict__ attribute
1454 return _convert_to_json_serializable_dict(obj.__dict__, visited, max_depth - 1)
1455 else:
1456 # Primitives (str, int, float, bool, None) pass through
1457 return obj
1458 finally:
1459 # Remove from visited set when done processing this object
1460 if obj_id in visited:
1461 visited.remove(obj_id)
1464def _convert_mapping_to_json_serializable(obj: Mapping[str, object]) -> dict[str, object]:
1465 converted: Final = _convert_to_json_serializable_dict(obj)
1466 if isinstance(converted, dict):
1467 return converted
1468 return dict(obj)
1471def _placeholder_stored_request_body_metadata(
1472 request_body: Mapping[str, object], persisted_model_group: str, raw_model: str
1473) -> Mapping[str, object]:
1474 body_metadata: Final = request_body.get("metadata")
1475 if not isinstance(body_metadata, Mapping):
1476 return request_body
1477 error_information: Final = body_metadata.get("error_information")
1478 placeholdered_fields: Final = MappingProxyType(
1479 {
1480 "model_group": persisted_model_group,
1481 "error_information": _scrub_raw_model_from_error_information(
1482 cast(StandardLoggingPayloadErrorInformation, error_information), raw_model
1483 )
1484 if isinstance(error_information, Mapping)
1485 else error_information,
1486 }
1487 )
1488 return MappingProxyType(
1489 {
1490 **request_body,
1491 "metadata": MappingProxyType(
1492 {key: placeholdered_fields.get(key, value) for key, value in body_metadata.items()}
1493 ),
1494 }
1495 )
1498def _placeholder_stored_request_body(
1499 litellm_params: Mapping[str, object], persisted_model_group: str, raw_model: str
1500) -> Mapping[str, object]:
1501 proxy_server_request: Final = litellm_params.get("proxy_server_request")
1502 if not isinstance(proxy_server_request, Mapping): 1502 ↛ 1503line 1502 didn't jump to line 1503 because the condition on line 1502 was never true
1503 return litellm_params
1504 request_body: Final = proxy_server_request.get("body")
1505 if not isinstance(request_body, Mapping):
1506 return litellm_params
1507 model_placeholdered: Final = (
1508 MappingProxyType({**request_body, "model": UNKNOWN_MODEL_SPEND_LOG_MODEL})
1509 if "model" in request_body
1510 else request_body
1511 )
1512 return MappingProxyType(
1513 {
1514 **litellm_params,
1515 "proxy_server_request": MappingProxyType(
1516 {
1517 **proxy_server_request,
1518 "body": _placeholder_stored_request_body_metadata(
1519 model_placeholdered, persisted_model_group, raw_model
1520 ),
1521 }
1522 ),
1523 }
1524 )
1527def _get_proxy_server_request_for_spend_logs_payload(
1528 metadata: dict,
1529 litellm_params: Mapping[str, object],
1530 kwargs: dict | None = None,
1531) -> str:
1532 """
1533 Only store if should_store_prompts_and_responses_in_spend_logs() is True
1535 If turn_off_message_logging is enabled, redact messages in the request body.
1536 """
1537 if should_store_prompts_and_responses_in_spend_logs(): 1537 ↛ 1538line 1537 didn't jump to line 1538 because the condition on line 1537 was never true
1538 _proxy_server_request: Final = cast(dict | None, litellm_params.get("proxy_server_request", EMPTY_MAPPING))
1539 if _proxy_server_request is not None:
1540 _request_body = _proxy_server_request.get("body", EMPTY_MAPPING) or EMPTY_MAPPING
1542 standard_payload: Final = (kwargs or EMPTY_MAPPING).get("standard_logging_object")
1543 if isinstance(standard_payload, Mapping):
1544 _request_body = MappingProxyType({**_request_body, **classifier_audit_fields(standard_payload)})
1546 if kwargs is not None:
1547 realtime_tools: Final = kwargs.get("realtime_tools")
1548 if realtime_tools:
1549 _request_body = dict(_request_body)
1550 _request_body["tools"] = realtime_tools
1552 # Apply message redaction if turn_off_message_logging is enabled
1553 if kwargs is not None:
1554 from litellm.litellm_core_utils.redact_messages import (
1555 perform_redaction,
1556 should_redact_message_logging,
1557 )
1559 # Build model_call_details dict to check redaction settings
1560 model_call_details: Final = {
1561 "litellm_params": litellm_params,
1562 "standard_callback_dynamic_params": kwargs.get("standard_callback_dynamic_params"),
1563 }
1565 # If redaction is enabled, convert to serializable dict before redacting
1566 if should_redact_message_logging(model_call_details=model_call_details):
1567 _request_body = _convert_mapping_to_json_serializable(without_classifier_audit(_request_body))
1568 perform_redaction(model_call_details=_request_body, result=None)
1570 _request_body = _sanitize_request_body_for_spend_logs_payload(_request_body)
1571 _request_body_json_str: Final = safe_dumps(_request_body)
1572 if LITELLM_TRUNCATED_PAYLOAD_FIELD in _request_body_json_str:
1573 verbose_proxy_logger.info(
1574 "Spend Log: request body was truncated before storing in DB. %s",
1575 LITELLM_TRUNCATION_DB_SAFEGUARD_NOTE,
1576 )
1577 return _request_body_json_str
1578 return "{}"
1581def _get_vector_store_request_for_spend_logs_payload(
1582 vector_store_request_metadata: list[StandardLoggingVectorStoreRequest] | None,
1583) -> list[StandardLoggingVectorStoreRequest] | None:
1584 """
1585 If user does not want to store prompts and responses, then remove the content from the vector store request metadata
1586 """
1587 if should_store_prompts_and_responses_in_spend_logs(): 1587 ↛ 1588line 1587 didn't jump to line 1588 because the condition on line 1587 was never true
1588 return vector_store_request_metadata
1590 # if user does not want to store prompts and responses, then remove the content from the vector store request metadata
1591 if vector_store_request_metadata is None: 1591 ↛ 1593line 1591 didn't jump to line 1593 because the condition on line 1591 was always true
1592 return None
1593 for vector_store_request in vector_store_request_metadata:
1594 vector_store_search_response: VectorStoreSearchResponse = (
1595 vector_store_request.get("vector_store_search_response") or VectorStoreSearchResponse()
1596 )
1597 response_data = vector_store_search_response.get("data", []) or []
1598 for response_item in response_data:
1599 for content_item in response_item.get("content", []) or []:
1600 if "text" in content_item:
1601 content_item["text"] = REDACTED_BY_LITELM_STRING
1602 return vector_store_request_metadata
1605def _get_response_for_spend_logs_payload(
1606 payload: StandardLoggingPayload | None,
1607 kwargs: dict | None = None,
1608) -> str:
1609 if payload is None:
1610 return "{}"
1611 if should_store_prompts_and_responses_in_spend_logs(): 1611 ↛ 1612line 1611 didn't jump to line 1612 because the condition on line 1611 was never true
1612 response_obj: object = payload.get("response")
1613 if response_obj is None:
1614 return "{}"
1616 if kwargs is not None:
1617 realtime_tool_calls: Final = kwargs.get("realtime_tool_calls")
1618 if realtime_tool_calls and isinstance(response_obj, dict):
1619 response_obj = dict(response_obj)
1620 response_obj["tool_calls"] = realtime_tool_calls
1622 # Apply message redaction if turn_off_message_logging is enabled
1623 if kwargs is not None:
1624 from litellm.litellm_core_utils.redact_messages import (
1625 perform_redaction,
1626 should_redact_message_logging,
1627 )
1629 litellm_params: Final = kwargs.get("litellm_params", {})
1630 model_call_details: Final = {
1631 "litellm_params": litellm_params,
1632 "standard_callback_dynamic_params": kwargs.get("standard_callback_dynamic_params"),
1633 }
1635 # If redaction is enabled, convert to serializable dict before redacting
1636 if should_redact_message_logging(model_call_details=model_call_details):
1637 response_obj = _convert_to_json_serializable_dict(response_obj)
1638 response_obj = perform_redaction(model_call_details={}, result=response_obj)
1640 sanitized_wrapper: Final = _sanitize_request_body_for_spend_logs_payload({"response": response_obj})
1642 sanitized_response: Final = sanitized_wrapper.get("response", response_obj)
1644 if sanitized_response is None:
1645 return "{}"
1646 if isinstance(sanitized_response, str):
1647 result_str = strip_null_bytes(sanitized_response)
1648 else:
1649 result_str = safe_dumps(sanitized_response)
1650 if LITELLM_TRUNCATED_PAYLOAD_FIELD in result_str:
1651 verbose_proxy_logger.info(
1652 "Spend Log: response was truncated before storing in DB. %s",
1653 LITELLM_TRUNCATION_DB_SAFEGUARD_NOTE,
1654 )
1655 return result_str
1656 return "{}"
1659def should_store_prompts_and_responses_in_spend_logs() -> bool:
1660 from litellm.proxy.proxy_server import general_settings
1661 from litellm.secret_managers.main import get_secret_bool
1663 # Check general_settings (from DB or proxy_config.yaml)
1664 store_prompts_value: Final = general_settings.get("store_prompts_in_spend_logs")
1666 # Normalize case: handle True/true/TRUE, False/false/FALSE, None/null
1667 if store_prompts_value is True: 1667 ↛ 1668line 1667 didn't jump to line 1668 because the condition on line 1667 was never true
1668 return True
1669 elif isinstance(store_prompts_value, str): 1669 ↛ 1671line 1669 didn't jump to line 1671 because the condition on line 1669 was never true
1670 # Case-insensitive string comparison
1671 if store_prompts_value.lower() == "true":
1672 return True
1674 # Also check environment variable
1675 return get_secret_bool("STORE_PROMPTS_IN_SPEND_LOGS") is True
1678def _get_status_for_spend_log(
1679 metadata: dict,
1680) -> Literal["success", "failure"]:
1681 """
1682 Get the status for the spend log.
1684 It's only a failure if metadata.get("status") is "failure"
1685 """
1686 _status: Final[str | None] = metadata.get("status", None)
1687 if _status == "failure":
1688 return "failure"
1689 return "success"