scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-10 22:41:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 81

refactor(types): replace Any with real types across 54 more backend files

Browse source 
Third batch of the fifth basedpyright Any reduction round. Every change is
typing-only and leaves runtime behavior identical.

Provider transformation configs, video and rerank base classes, OTel metadata
and the guardrail and realtime type modules move their payload, header and
optional-parameter annotations from Any to object, Mapping[str, object] or the
concrete model the call site already produces. Repositories and endpoints that
reached Prisma through an untyped handle now name the actions they call with the
repo's own TableActions protocol. The pydantic field retypes were checked
against pydantic to confirm object and Any validate, serialize and generate JSON
schema identically.
This commit is contained in:
mateo-berri 2026-09-08 12:41:07 +00:00
parent f747bc6704
commit 1ec5083ab4
54 changed files with 323 additions and 202 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
litellm/a2a_protocol/providers/pydantic_ai_agents/config.py
View file

@ -23,7 +23,7 @@ class PydanticAIProviderConfig(BaseA2AProviderConfig):
params: dict[str, Any],
api_base: str | None = None,
**kwargs: Any,
) -> dict[str, Any]:
) -> dict[str, object]:
"""Handle non-streaming request to Pydantic AI agent."""
if api_base is None:
raise ValueError("api_base is required for PydanticAIProviderConfig")
@ -41,7 +41,7 @@ class PydanticAIProviderConfig(BaseA2AProviderConfig):
params: dict[str, Any],
api_base: str | None = None,
**kwargs,
) -> AsyncIterator[dict[str, Any]]:
) -> AsyncIterator[dict[str, object]]:
"""Handle streaming request with fake streaming."""
if not api_base:
raise ValueError("api_base is required for Pydantic AI agents")

7
litellm/integrations/argilla.py
View file

@ -7,7 +7,8 @@ import json
import os
import random
import types
from typing import Any, Final
from collections.abc import Mapping
from typing import Final
import httpx
from pydantic import BaseModel
@ -69,7 +70,7 @@ class ArgillaLogger(CustomBatchLogger):
self.flush_lock = asyncio.Lock()
super().__init__(**kwargs, flush_lock=self.flush_lock)
def validate_argilla_transformation_object(self, argilla_transformation_object: dict[str, Any]):
def validate_argilla_transformation_object(self, argilla_transformation_object: Mapping[str, object]):
if not isinstance(argilla_transformation_object, dict):
raise Exception("'argilla_transformation_object' must be a dictionary, to log your payload to Argilla.")
@ -115,7 +116,7 @@ class ArgillaLogger(CustomBatchLogger):
ARGILLA_DATASET_NAME=_credentials_dataset_name,
)
def get_chat_messages(self, payload: StandardLoggingPayload) -> list[dict[str, Any]]:
def get_chat_messages(self, payload: StandardLoggingPayload) -> list[dict[str, object]]:
payload_messages: Final = payload.get("messages", None)
if payload_messages is None:

19
litellm/integrations/dynamodb.py
View file

@ -3,12 +3,25 @@
import os
import traceback
from typing import Any, Final
from collections.abc import Mapping
from typing import Final, Protocol
import litellm
from litellm._uuid import uuid
class _DynamoTable(Protocol):
"""The one boto3 DynamoDB table call this logger makes."""
def put_item(self, *, Item: Mapping[str, object]) -> object: ...
class _DynamoResource(Protocol):
"""The one boto3 DynamoDB resource call this logger makes."""
def Table(self, name: str) -> _DynamoTable: ...
class DyanmoDBLogger:
# Class variables or attributes
@ -16,7 +29,7 @@ class DyanmoDBLogger:
# Instance variables
import boto3
self.dynamodb: Any = boto3.resource("dynamodb", region_name=os.environ["AWS_REGION_NAME"])
self.dynamodb: Final[_DynamoResource] = boto3.resource("dynamodb", region_name=os.environ["AWS_REGION_NAME"])
if litellm.dynamodb_table_name is None:
raise ValueError(
"LiteLLM Error, trying to use DynamoDB but not table name passed. Create a table and set `litellm.dynamodb_table_name=<your-table>`"
@ -41,7 +54,7 @@ class DyanmoDBLogger:
id: Final = response_obj.get("id", str(uuid.uuid4()))
# Build the initial payload
payload: Final = {
payload: Final[dict[str, object]] = {
"id": id,
"call_type": call_type,
"startTime": start_time,

6
litellm/integrations/focus/database.py
View file

@ -3,7 +3,7 @@
from __future__ import annotations
from datetime import datetime
from typing import Any, Final
from typing import Final
import polars as pl
@ -32,7 +32,7 @@ class FocusLiteLLMDatabase:
client: Final = self._ensure_prisma_client()
where_clauses: Final[list[str]] = []
query_params: Final[list[Any]] = []
query_params: Final[list[datetime | int]] = []
placeholder_index = 1
if start_time_utc:
where_clauses.append(f"dus.updated_at >= ${placeholder_index}::timestamptz")
@ -112,7 +112,7 @@ class FocusLiteLLMDatabase:
except Exception as exc:
raise RuntimeError(f"Error retrieving usage data: {exc}") from exc
async def get_table_info(self) -> dict[str, Any]:
async def get_table_info(self) -> dict[str, object]:
"""Return metadata about the spend table for diagnostics."""
client: Final = self._ensure_prisma_client()

5
litellm/integrations/focus/destinations/vantage_destination.py
View file

@ -4,7 +4,8 @@ from __future__ import annotations
import csv
import io
from typing import Any, Final
from collections.abc import Mapping
from typing import Final
import httpx # noqa: F401 - used at runtime (AsyncClient, HTTPStatusError)
@ -94,7 +95,7 @@ class FocusVantageDestination(FocusDestination):
self,
*,
prefix: str,
config: dict[str, Any] | None = None,
config: Mapping[str, object] | None = None,
) -> None:
config = config or {}
api_key: Final = config.get("api_key")

3
litellm/integrations/langfuse/langfuse_otel_attributes.py
View file

@ -5,6 +5,7 @@ Relevant Issue: https://github.com/BerriAI/litellm/issues/13764
"""
import json
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
from pydantic import BaseModel
@ -40,7 +41,7 @@ def get_output_content_by_type(
| HttpxBinaryResponseContent
| ResponsesAPIResponse
| list,
kwargs: dict[str, Any] | None = None,
kwargs: Mapping[str, object] | None = None,
) -> str:
"""
Extract output content from response objects based on their type.

8
litellm/integrations/langsmith.py
View file

@ -75,9 +75,9 @@ class LangsmithLogger(CustomBatchLogger):
if _batch_size:
self.batch_size = int(_batch_size)
self.log_queue: list[LangsmithQueueObject] = []
self._flush_task: asyncio.Task[Any] | None = self._start_periodic_flush_task()
self._flush_task: asyncio.Task[None] | None = self._start_periodic_flush_task()
def _start_periodic_flush_task(self) -> asyncio.Task[Any] | None:
def _start_periodic_flush_task(self) -> asyncio.Task[None] | None:
"""Start the periodic flush task only when an event loop is already running."""
try:
loop: Final = asyncio.get_running_loop()
@ -152,9 +152,9 @@ class LangsmithLogger(CustomBatchLogger):
return self._redact_metadata(extra_metadata)
def _build_outputs_with_usage(self, payload: StandardLoggingPayload) -> dict[str, Any]:
def _build_outputs_with_usage(self, payload: StandardLoggingPayload) -> dict[str, object]:
response: Final = payload["response"]
outputs: dict[str, Any]
outputs: dict[str, object]
if isinstance(response, dict):
outputs = {**response}
else:

14
litellm/integrations/otel/model/metadata.py
View file

@ -36,7 +36,7 @@ model. They coincide on the SDK path, which is correct.
from __future__ import annotations
from collections.abc import Iterator, Mapping
from collections.abc import Callable, Iterator, Mapping
from dataclasses import dataclass, field
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, cast
@ -57,7 +57,7 @@ class RequestIdentity:
# The team's free-form metadata, carried raw (empty/missing -> None) and
# filtered to an operator allowlist only at Baggage-promotion time, so an
# unconfigured deployment never promotes any of it.
team_metadata: Mapping[str, Any] | None = None
team_metadata: Mapping[str, object] | None = None
key_hash: str | None = None
end_user: str | None = None
# The model litellm dispatched to the provider. Only known once the call
@ -103,7 +103,7 @@ class RequestIdentity:
``user_api_key_*`` names that ``baggage.DEFAULT_BAGGAGE_METADATA_KEYS``
promotes.
"""
get: Final = lambda name: getattr(auth, name, None) # noqa: E731
get: Final[Callable[[str], object]] = lambda name: getattr(auth, name, None) # noqa: E731
metadata: Final = {
meta_key: str(value)
for meta_key, attr in (
@ -217,7 +217,7 @@ class LLMCallEvent:
time_to_first_chunk_seconds: float | None
@classmethod
def from_dict(cls, kwargs: Mapping[str, Any]) -> LLMCallEvent:
def from_dict(cls, kwargs: Mapping[str, object]) -> LLMCallEvent:
raw_payload: Final = kwargs.get("standard_logging_object")
payload: Final = cast("StandardLoggingPayload", raw_payload) if raw_payload else None
operation: Final = resolve_operation(as_str(kwargs.get("call_type")))
@ -239,7 +239,7 @@ def time_to_first_chunk_seconds(kwargs: Mapping[str, Any]) -> float | None:
to the first streamed chunk (``completion_start_time``); ``None`` for
non-streaming calls, where ``completion_start_time`` is backfilled with the
end time and would not measure first-chunk latency."""
optional_params: Final = cast(Mapping[str, Any], kwargs.get("optional_params") or {})
optional_params: Final = cast(Mapping[str, object], kwargs.get("optional_params") or {})
if not optional_params.get("stream"):
return None
api_call_start: Final = to_seconds(kwargs.get("api_call_start_time"))
@ -307,7 +307,7 @@ def _metadata_dicts(
)
def _call_id(payload: StandardLoggingPayload | None, kwargs: Mapping[str, Any]) -> str | None:
def _call_id(payload: StandardLoggingPayload | None, kwargs: Mapping[str, object]) -> str | None:
"""The call id from the payload (when closed) or the bare kwargs (at pre_call)."""
if payload is not None:
call_id: Final = as_str(payload.get("litellm_call_id")) or as_str(payload.get("id"))
@ -351,7 +351,7 @@ def _model_info_id(model_info: object) -> str | None:
return None
def _team_metadata_dict(value: object) -> Mapping[str, Any] | None:
def _team_metadata_dict(value: object) -> Mapping[str, object] | None:
"""The team's free-form metadata as a raw mapping, or ``None`` when missing
or empty.

19
litellm/integrations/otel/mount.py
View file

@ -12,11 +12,14 @@ when the feature gate is off.
"""
import os
from typing import Any, Final
from typing import TYPE_CHECKING, Final, Protocol
from litellm._logging import verbose_logger
from litellm.integrations.otel.model.config import is_otel_v2_enabled
if TYPE_CHECKING:
from fastapi import FastAPI
# Routes excluded from server-span tracing by default: high-frequency pollers and
# static UI/docs assets, none of which are LLM traffic. Entries are substring-matched
# against the request path (unanchored, so they survive a ``server_root_path`` prefix
@ -65,7 +68,17 @@ PASSTHROUGH_PREFIXES: Final = frozenset(
)
def _passthrough_span_name_hook(span: Any, scope: dict) -> None:
class _RenameableSpan(Protocol):
"""The span surface the passthrough naming hook drives."""
def is_recording(self) -> bool: ...
def update_name(self, name: str) -> None: ...
def set_attribute(self, key: str, value: str) -> None: ...
def _passthrough_span_name_hook(span: "_RenameableSpan | None", scope: dict) -> None:
"""FastAPI ``server_request_hook``: give passthrough server spans a useful name.
The instrumentation matches the route at span creation, so both the span name
@ -88,7 +101,7 @@ def _passthrough_span_name_hook(span: Any, scope: dict) -> None:
pass
def instrument_fastapi_app(app: Any) -> None:
def instrument_fastapi_app(app: "FastAPI") -> None:
"""Attach OTel server-span instrumentation to the proxy FastAPI app.
Safe no-op when the V2 gate is off or ``opentelemetry-instrumentation-fastapi``

6
litellm/litellm_core_utils/coroutine_checker.py
View file

@ -16,7 +16,7 @@ class CoroutineChecker:
"""
def __init__(self):
self._cache = WeakKeyDictionary()
self._cache: WeakKeyDictionary[object, bool] = WeakKeyDictionary()
self._max_size = COROUTINE_CHECKER_MAX_SIZE_IN_MEMORY
def is_async_callable(self, callback: Any) -> bool:
@ -33,10 +33,10 @@ class CoroutineChecker:
pass
# Determine target - optimized path for common cases
target = callback
target: object = callback
if not inspect.isfunction(target) and not inspect.ismethod(target):
try:
call_attr: Final = getattr(target, "__call__", None)
call_attr: Final[object] = getattr(target, "__call__", None)
if call_attr is not None:
target = call_attr
except Exception:

6
litellm/litellm_core_utils/exception_mapping_utils.py
View file

@ -1,7 +1,7 @@
import json
import re
import traceback
from typing import Any, Final, Protocol, cast
from typing import Final, Protocol, cast
import httpx
@ -191,7 +191,7 @@ def _get_response_headers(original_exception: Exception) -> httpx.Headers | None
_response_headers: httpx.Headers | None = None
try:
_response_headers = getattr(original_exception, "headers", None)
error_response: Final = getattr(original_exception, "response", None)
error_response: Final[object] = getattr(original_exception, "response", None)
if not _response_headers and error_response:
_response_headers = getattr(error_response, "headers", None)
if not _response_headers:
@ -203,7 +203,7 @@ def _get_response_headers(original_exception: Exception) -> httpx.Headers | None
def extract_and_raise_litellm_exception(
response: Any | None,
response: object | None,
error_str: str,
model: str,
custom_llm_provider: str,

26
litellm/litellm_core_utils/prompt_templates/huggingface_template_handler.py
View file

@ -1,6 +1,8 @@
import json
from datetime import datetime
from typing import Any, Final
from typing import Any, Final, Literal
from typing_extensions import NotRequired, ReadOnly, TypedDict
from litellm.llms.custom_httpx.http_handler import (
_get_httpx_client,
@ -9,6 +11,20 @@ from litellm.llms.custom_httpx.http_handler import (
from litellm.types.llms.custom_http import httpxSpecialProvider
class _TokenizerConfigResult(TypedDict):
"""Outcome of a tokenizer_config.json fetch, carrying the parsed document when the fetch succeeded."""
status: ReadOnly[Literal["success", "failure"]]
tokenizer: NotRequired[ReadOnly[object]]
class _ChatTemplateFileResult(TypedDict):
"""Outcome of a chat template file fetch, carrying the template body when the fetch succeeded."""
status: ReadOnly[Literal["success", "failure"]]
chat_template: NotRequired[ReadOnly[str]]
def strftime_now(fmt: str) -> str:
"""
Custom function for templates that need current date/time formatting (e.g., gpt-oss)
@ -22,7 +38,7 @@ def strftime_now(fmt: str) -> str:
return datetime.now().strftime(fmt)
def _get_tokenizer_config(hf_model_name: str) -> dict[str, Any]:
def _get_tokenizer_config(hf_model_name: str) -> _TokenizerConfigResult:
"""
Fetch tokenizer_config.json from HuggingFace (sync)
@ -45,7 +61,7 @@ def _get_tokenizer_config(hf_model_name: str) -> dict[str, Any]:
return {"status": "failure"}
async def _aget_tokenizer_config(hf_model_name: str) -> dict[str, Any]:
async def _aget_tokenizer_config(hf_model_name: str) -> _TokenizerConfigResult:
"""
Fetch tokenizer_config.json from HuggingFace (async)
@ -70,7 +86,7 @@ async def _aget_tokenizer_config(hf_model_name: str) -> dict[str, Any]:
return {"status": "failure"}
def _get_chat_template_file(hf_model_name: str) -> dict[str, Any]:
def _get_chat_template_file(hf_model_name: str) -> _ChatTemplateFileResult:
"""
Fetch chat template from separate .jinja file (sync)
@ -98,7 +114,7 @@ def _get_chat_template_file(hf_model_name: str) -> dict[str, Any]:
return {"status": "failure"}
async def _aget_chat_template_file(hf_model_name: str) -> dict[str, Any]:
async def _aget_chat_template_file(hf_model_name: str) -> _ChatTemplateFileResult:
"""
Fetch chat template from separate .jinja file (async)

2
litellm/llms/azure_ai/anthropic/count_tokens/handler.py
View file

@ -33,7 +33,7 @@ class AzureAIAnthropicCountTokensHandler(AzureAIAnthropicCountTokensConfig):
litellm_params: dict[str, Any] | None = None,
timeout: float | httpx.Timeout | None = None,
tools: list[dict[str, Any]] | None = None,
system: Any | None = None,
system: object = None,
) -> dict[str, Any]:
"""
Handle a CountTokens request using httpx with Azure authentication.

10
litellm/llms/base_llm/videos/transformation.py
View file

@ -180,7 +180,7 @@ class BaseVideoConfig(ABC):
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
extra_body: dict[str, Any] | None = None,
extra_body: dict[str, object] | None = None,
) -> tuple[str, dict]:
"""
Transform the video remix request into a URL and data
@ -207,7 +207,7 @@ class BaseVideoConfig(ABC):
after: str | None = None,
limit: int | None = None,
order: str | None = None,
extra_query: dict[str, Any] | None = None,
extra_query: dict[str, object] | None = None,
) -> tuple[str, dict]:
"""
Transform the video list request into a URL and params
@ -342,8 +342,8 @@ class BaseVideoConfig(ABC):
litellm_params: GenericLiteLLMParams,
headers: dict,
video_file: FileContent | None = None,
extra_body: dict[str, Any] | None = None,
prefetched_source_data: dict[str, Any] | None = None,
extra_body: dict[str, object] | None = None,
prefetched_source_data: dict[str, object] | None = None,
) -> tuple[str, Mapping[str, object], RequestFiles | None]:
"""
Transform the video edit request into a URL plus either JSON data or
@ -373,7 +373,7 @@ class BaseVideoConfig(ABC):
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
extra_body: dict[str, Any] | None = None,
extra_body: dict[str, object] | None = None,
) -> tuple[str, dict]:
"""
Transform the video extension request into a URL and JSON data.

11
litellm/llms/bedrock/count_tokens/bedrock_token_counter.py
View file

@ -2,6 +2,7 @@
Bedrock Token Counter implementation using the CountTokens API.
"""
from collections.abc import Mapping, Sequence
from typing import Any, Final
from litellm._logging import verbose_logger
@ -26,12 +27,12 @@ class BedrockTokenCounter(BaseTokenCounter):
async def count_tokens(
self,
model_to_use: str,
messages: list[dict[str, Any]] | None,
contents: list[dict[str, Any]] | None,
messages: Sequence[Mapping[str, object]] | None,
contents: Sequence[Mapping[str, object]] | None,
deployment: dict[str, Any] | None = None,
request_model: str = "",
tools: list[dict[str, Any]] | None = None,
system: Any | None = None,
tools: Sequence[Mapping[str, object]] | None = None,
system: object | None = None,
) -> TokenCountResponse | None:
"""
Count tokens using AWS Bedrock's CountTokens API.
@ -56,7 +57,7 @@ class BedrockTokenCounter(BaseTokenCounter):
litellm_params: Final = deployment.get("litellm_params", {})
# Build request data in the format expected by BedrockCountTokensHandler
request_data: Final[dict[str, Any]] = {
request_data: Final[dict[str, object]] = {
"model": model_to_use,
"messages": messages,
}

6
litellm/llms/bedrock/files/transformation.py
View file

@ -247,7 +247,7 @@ def _validate_file_id_against_configured_buckets(
return validate_against(configured_bucket_names[-1])
def _uploaded_object_size(litellm_params: Mapping[str, object], raw_response: Response) -> int:
def _uploaded_object_size(litellm_params: Mapping[str, object], response_headers: Mapping[str, str]) -> int:
"""
S3 answers PutObject with an empty body, so the stored object size comes from the
signed request recorded by `transform_create_file_request`, not the response headers.
@ -255,7 +255,7 @@ def _uploaded_object_size(litellm_params: Mapping[str, object], raw_response: Re
uploaded_size: Final = litellm_params.get(UPLOAD_CONTENT_LENGTH_PARAM)
if isinstance(uploaded_size, int):
return uploaded_size
response_content_length: Final = raw_response.headers.get("Content-Length", "0")
response_content_length: Final = response_headers.get("Content-Length", "0")
return int(response_content_length) if response_content_length.isdigit() else 0
@ -1161,7 +1161,7 @@ class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig):
filename=filename,
created_at=int(time.time()), # Current timestamp
status="uploaded",
bytes=_uploaded_object_size(litellm_params=litellm_params, raw_response=raw_response),
bytes=_uploaded_object_size(litellm_params=litellm_params, response_headers=raw_response.headers),
object="file",
)

5
litellm/llms/bedrock/passthrough/guardrail_translation/handler.py
View file

@ -14,6 +14,9 @@ from litellm.types.utils import GenericGuardrailAPIInputs
if TYPE_CHECKING:
from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.pass_through.guardrail_translation.handler import (
PassThroughEndpointHandler,
)
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.utils import ProxyLogging
@ -27,7 +30,7 @@ def _is_converse_endpoint(endpoint: str) -> bool:
return bool(parts) and parts[-1] in _CONVERSE_ACTIONS
def _generic_passthrough_handler() -> BaseTranslation:
def _generic_passthrough_handler() -> "PassThroughEndpointHandler":
"""
Fallback for non-Converse Bedrock routes (e.g. invoke). The generic
handler scans the full request/response payload so blocking guardrails

4
litellm/llms/chatgpt/common_utils.py
View file

@ -268,7 +268,7 @@ def _normalize_litellm_params(litellm_params: Any | None) -> dict:
return {}
def get_chatgpt_session_id(litellm_params: Any | None) -> str | None:
def get_chatgpt_session_id(litellm_params: object) -> str | None:
params: Final = _normalize_litellm_params(litellm_params)
for key in ("litellm_session_id", "session_id"):
value = params.get(key)
@ -286,5 +286,5 @@ def get_chatgpt_session_id(litellm_params: Any | None) -> str | None:
return None
def ensure_chatgpt_session_id(litellm_params: Any | None) -> str:
def ensure_chatgpt_session_id(litellm_params: object) -> str:
return get_chatgpt_session_id(litellm_params) or str(uuid4())

7
litellm/llms/github_copilot/responses/transformation.py
View file

@ -19,6 +19,7 @@ from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfi
from litellm.types.llms.openai import (
ResponseInputParam,
ResponsesAPIOptionalRequestParams,
ResponsesAPIStreamingResponse,
)
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import LlmProviders
@ -129,7 +130,7 @@ class GithubCopilotResponsesAPIConfig(OpenAIResponsesAPIConfig):
model: str,
parsed_chunk: dict,
logging_obj: LiteLLMLoggingObj,
) -> Any:
) -> ResponsesAPIStreamingResponse:
parsed_chunk = self._normalize_stream_item_id(parsed_chunk)
return super().transform_streaming_response(
model=model,
@ -262,7 +263,7 @@ class GithubCopilotResponsesAPIConfig(OpenAIResponsesAPIConfig):
# Return the responses endpoint
return f"{effective_api_base}/responses"
def _handle_reasoning_item(self, item: dict[str, Any]) -> dict[str, Any]:
def _handle_reasoning_item(self, item: dict[str, object]) -> dict[str, object]:
"""
Handle reasoning items for GitHub Copilot, preserving encrypted_content.
@ -280,7 +281,7 @@ class GithubCopilotResponsesAPIConfig(OpenAIResponsesAPIConfig):
# Filter out None values for known problematic fields,
# but preserve encrypted_content even if it exists
filtered_item: Final[dict[str, Any]] = {}
filtered_item: Final[dict[str, object]] = {}
for k, v in item.items():
# Always include encrypted_content if present (even if None)
if k == "encrypted_content":

21
litellm/llms/hosted_vllm/chat/transformation.py
View file

@ -28,12 +28,12 @@ from ...openai.chat.gpt_transformation import OpenAIGPTConfig
class HostedVLLMChatConfig(OpenAIGPTConfig):
def _convert_custom_tools_to_function_tools(self, tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
def _convert_custom_tools_to_function_tools(self, tools: list[dict[str, Any]]) -> list[dict[str, object]]:
"""
vLLM chat completions currently accepts only OpenAI function tools.
Convert custom tools into function tools so request validation does not fail.
"""
converted_tools: Final[list[dict[str, Any]]] = []
converted_tools: Final[list[dict[str, object]]] = []
for idx, tool in enumerate(tools):
if not isinstance(tool, dict):
converted_tools.append(tool)
@ -63,17 +63,14 @@ class HostedVLLMChatConfig(OpenAIGPTConfig):
"required": ["input"],
}
function_tool: dict[str, Any] = {
"type": "function",
"function": {
"name": str(tool_name),
"parameters": tool_parameters,
},
function_definition: dict[str, object] = {
"name": str(tool_name),
"parameters": tool_parameters,
}
if isinstance(tool_description, str):
function_tool["function"]["description"] = tool_description
function_definition["description"] = tool_description
converted_tools.append(function_tool)
converted_tools.append({"type": "function", "function": function_definition})
return converted_tools
@ -148,7 +145,7 @@ class HostedVLLMChatConfig(OpenAIGPTConfig):
@overload
def _transform_messages(
self, messages: list[AllMessageValues], model: str, is_async: Literal[True]
) -> Coroutine[Any, Any, list[AllMessageValues]]: ...
) -> Coroutine[object, object, list[AllMessageValues]]: ...
@overload
def _transform_messages(
@ -160,7 +157,7 @@ class HostedVLLMChatConfig(OpenAIGPTConfig):
def _transform_messages(
self, messages: list[AllMessageValues], model: str, is_async: bool = False
) -> list[AllMessageValues] | Coroutine[Any, Any, list[AllMessageValues]]:
) -> list[AllMessageValues] | Coroutine[object, object, list[AllMessageValues]]:
"""
Support translating:
- video files from file_id or file_data to video_url

6
litellm/llms/openai/common_utils.py
View file

@ -10,7 +10,7 @@ import ssl
import time
import uuid
from collections.abc import AsyncIterator, Iterator, Mapping
from typing import TYPE_CHECKING, Any, Final, Literal, NamedTuple, Optional
from typing import TYPE_CHECKING, Final, Literal, NamedTuple, Optional
from urllib.parse import urlsplit
import httpx
@ -87,8 +87,8 @@ class OpenAIError(BaseLLMException):
###################################################################
def drop_params_from_unprocessable_entity_error(
e: openai.UnprocessableEntityError | httpx.HTTPStatusError,
data: dict[str, Any],
) -> dict[str, Any]:
data: Mapping[str, object],
) -> dict[str, object]:
"""
Helper function to read OpenAI UnprocessableEntityError and drop the params that raised an error from the error message.

4
litellm/llms/snowflake/chat/transformation.py
View file

@ -315,7 +315,7 @@ class SnowflakeConfig(SnowflakeBaseConfig, OpenAIGPTConfig):
for msg in messages:
if isinstance(msg, dict):
role = msg.get("role", "")
content: Any = msg.get("content", "")
content: object = msg.get("content", "")
msg_cache_control: object = msg.get("cache_control")
else:
role = getattr(msg, "role", "")
@ -463,7 +463,7 @@ class SnowflakeConfig(SnowflakeBaseConfig, OpenAIGPTConfig):
return body
def _transform_tool_choice_to_anthropic(self, tool_choice: Any) -> dict[str, Any]:
def _transform_tool_choice_to_anthropic(self, tool_choice: object) -> Mapping[str, object]:
"""
Convert tool_choice from OpenAI format to Anthropic format.

4
litellm/llms/stability/image_edit/transformations.py
View file

@ -74,7 +74,7 @@ class StabilityImageEditConfig(BaseImageEditConfig):
}
# Create a copy to not mutate original - convert TypedDict to regular dict
mapped_params: Final[dict[str, Any]] = dict(image_edit_optional_params)
mapped_params: Final[dict[str, object]] = dict(image_edit_optional_params)
for k, v in image_edit_optional_params.items():
if k in param_mapping:
@ -182,7 +182,7 @@ class StabilityImageEditConfig(BaseImageEditConfig):
# Build Stability request
# Populate multipart form-data as separate text fields (data) and files.
# Stability expects prompt/output_format/etc. as normal form fields, not file parts.
data: Final[dict[str, Any]] = {
data: Final[dict[str, object]] = {
"output_format": "png", # Default to PNG
}

8
litellm/llms/triton/completion/transformation.py
View file

@ -4,7 +4,7 @@ Translates from OpenAI's `/v1/chat/completions` endpoint to Triton's `/generate`
import json
from collections.abc import AsyncIterator, Iterator
from typing import TYPE_CHECKING, Any, Final, Literal
from typing import TYPE_CHECKING, Final, Literal
from httpx import Headers, Response
@ -172,7 +172,7 @@ class TritonConfig(BaseConfig):
streaming_response: Iterator[str] | AsyncIterator[str] | ModelResponse,
sync_stream: bool,
json_mode: bool | None = False,
) -> Any:
) -> "TritonResponseIterator":
return TritonResponseIterator(
streaming_response=streaming_response,
sync_stream=sync_stream,
@ -195,14 +195,14 @@ class TritonGenerateConfig(TritonConfig):
) -> dict:
inference_params: Final = optional_params.copy()
stream: Final = inference_params.pop("stream", False)
data_for_triton: Final[dict[str, Any]] = {
data_for_triton: Final[dict[str, object]] = {
"text_input": prompt_factory(model=model, messages=messages),
"parameters": {
"max_tokens": int(optional_params.get("max_tokens", DEFAULT_MAX_TOKENS_FOR_TRITON)),
**inference_params,
},
"stream": bool(stream),
}
data_for_triton["parameters"].update(inference_params)
return data_for_triton
def transform_response(

4
litellm/llms/vertex_ai/rerank/transformation.py
View file

@ -5,7 +5,7 @@ Why separate file? Make it easy to see how transformation works
"""
from collections.abc import Mapping
from typing import Any, Final
from typing import Final
import httpx
@ -227,7 +227,7 @@ class VertexAIRerankConfig(BaseRerankConfig, VertexBase):
model: str,
drop_params: bool,
query: str,
documents: list[str | dict[str, Any]],
documents: list[str | dict[str, object]],
custom_llm_provider: str | None = None,
top_n: int | None = None,
rank_fields: list[str] | None = None,

9
litellm/llms/volcengine/embedding/transformation.py
View file

@ -3,7 +3,8 @@ Volcengine Embedding Transformation
Transforms OpenAI embedding requests to Volcengine format
"""
from typing import Any, Final
from collections.abc import Mapping
from typing import Final
import httpx
@ -83,11 +84,11 @@ class VolcEngineEmbeddingConfig(BaseEmbeddingConfig):
def map_openai_params(
self,
non_default_params: dict[str, Any],
optional_params: dict[str, Any],
non_default_params: Mapping[str, object],
optional_params: dict[str, object],
model: str,
drop_params: bool,
) -> dict[str, Any]:
) -> dict[str, object]:
"""
Map OpenAI embedding parameters to Volcengine format.

8
litellm/llms/watsonx/rerank/transformation.py
View file

@ -5,8 +5,8 @@ Docs - https://cloud.ibm.com/apidocs/watsonx-ai#text-rerank
"""
import uuid
from collections.abc import Mapping
from typing import Any, Final, cast
from collections.abc import Mapping, Sequence
from typing import Final, cast
import httpx
@ -96,7 +96,7 @@ class IBMWatsonXRerankConfig(IBMWatsonXMixin, BaseRerankConfig):
model: str,
drop_params: bool,
query: str,
documents: list[str | dict[str, Any]],
documents: Sequence[str | Mapping[str, object]],
custom_llm_provider: str | None = None,
top_n: int | None = None,
rank_fields: list[str] | None = None,
@ -178,7 +178,7 @@ class IBMWatsonXRerankConfig(IBMWatsonXMixin, BaseRerankConfig):
transformed_results: Final = []
for result in _results:
transformed_result: dict[str, Any] = {
transformed_result: dict[str, object] = {
"index": result["index"],
"relevance_score": result["score"],
}

18
litellm/llms/xai/realtime/transformation.py
View file

@ -16,7 +16,7 @@ construction time (see ``handler.py``) so all normalization is isolated here
and ``RealTimeStreaming`` stays provider-agnostic.
"""
from typing import Any, Final
from typing import Final
class XAIRealtimeNormalizer:
@ -58,7 +58,7 @@ class XAIRealtimeNormalizer:
# Cache content-part objects keyed by (response_id, item_id, content_index)
# so that ``response.content_part.done`` events missing ``part`` can be
# back-filled from earlier ``content_part.added`` / delta-done events.
self._content_part_by_key: dict[tuple, dict[str, Any]] = {}
self._content_part_by_key: dict[tuple, dict[str, object]] = {}
# ---------------------------------------------------------------------------
# Public interface consumed by RealTimeStreaming
@ -140,7 +140,7 @@ class XAIRealtimeNormalizer:
}
self._content_part_by_key[key] = updated
def _resolve_content_part(self, event: dict) -> dict[str, Any]:
def _resolve_content_part(self, event: dict) -> dict[str, object]:
part: Final = event.get("part")
if isinstance(part, dict):
return part
@ -214,7 +214,7 @@ class XAIRealtimeNormalizer:
needs_content: Final = event_type in self._EVENTS_NEEDING_CONTENT_INDEX
if not needs_output and not needs_content:
return event
patch: Final[dict[str, Any]] = {}
patch: Final[dict[str, object]] = {}
if needs_output and "output_index" not in event:
patch["output_index"] = 0
if needs_content and "content_index" not in event:
@ -228,8 +228,8 @@ class XAIRealtimeNormalizer:
# ---------------------------------------------------------------------------
@staticmethod
def _default_ga_usage() -> dict[str, Any]:
default_details: Final[dict[str, Any]] = {
def _default_ga_usage() -> dict[str, object]:
default_details: Final[dict[str, int]] = {
"cached_tokens": 0,
"text_tokens": 0,
"audio_tokens": 0,
@ -243,7 +243,7 @@ class XAIRealtimeNormalizer:
}
@staticmethod
def _normalize_usage(usage: object, *, empty_as_null: bool) -> dict[str, Any] | None:
def _normalize_usage(usage: object, *, empty_as_null: bool) -> dict[str, object] | None:
"""Coerce a usage object into the full OpenAI GA shape.
``empty_as_null=True`` for ``response.created`` (usage optional).
@ -253,12 +253,12 @@ class XAIRealtimeNormalizer:
return None
if not usage:
return None if empty_as_null else XAIRealtimeNormalizer._default_ga_usage()
default_details: Final[dict[str, Any]] = {
default_details: Final[dict[str, int]] = {
"cached_tokens": 0,
"text_tokens": 0,
"audio_tokens": 0,
}
normalized: Final[dict[str, Any]] = {
normalized: Final[dict[str, object]] = {
"total_tokens": usage.get("total_tokens", 0),
"input_tokens": usage.get("input_tokens", 0),
"output_tokens": usage.get("output_tokens", 0),

20
litellm/proxy/analytics_endpoints/cache_activity.py
View file

@ -2,16 +2,26 @@ import asyncio
import json
from collections.abc import Sequence
from datetime import datetime
from typing import TYPE_CHECKING, Final
from typing import Final, Protocol
from pydantic import BaseModel, TypeAdapter
if TYPE_CHECKING:
from litellm.proxy.utils import PrismaClient
UNKNOWN_CALL_TYPE: Final = "Unknown"
class _SupportsQueryRaw(Protocol):
"""The single database operation the cache-activity queries issue."""
async def query_raw(self, query: str, *args: object) -> Sequence[object]: ...
class _SupportsRawQueryDb(Protocol):
"""A prisma client handle, narrowed to the raw-query surface used here."""
@property
def db(self) -> _SupportsQueryRaw: ...
class CacheActivityGroup(BaseModel):
call_type: str
api_requests: int
@ -143,7 +153,7 @@ def compute_totals(groups: Sequence[CacheActivityGroup]) -> CacheActivityTotals:
async def get_cache_activity(
prisma_client: "PrismaClient",
prisma_client: _SupportsRawQueryDb,
start_date: datetime,
end_date: datetime,
key_aliases: Sequence[str],

18
litellm/proxy/client/cli/commands/model_groups.py
View file

@ -1,3 +1,4 @@
from collections.abc import Mapping
from typing import Final, Literal
import click
@ -5,10 +6,17 @@ import rich
import rich.table
from ... import Client
from ._cli_context import cli_context_values
def create_client(ctx: click.Context) -> Client:
return Client(base_url=ctx.obj["base_url"], api_key=ctx.obj["api_key"])
context: Final = cli_context_values(ctx)
return Client(base_url=context["base_url"], api_key=context["api_key"])
def _rendered_field(group: Mapping[str, object], key: str, default: str) -> str:
"""The rendered value of one model group field, or ``default`` when the group omits it."""
return str(group.get(key, default))
@click.group(name="model-groups")
@ -46,10 +54,10 @@ def list_model_groups(ctx: click.Context, output_format: Literal["table", "json"
for group in groups:
table.add_row(
str(group.get("model_group", "")),
str(group.get("mode", "chat")),
str(group.get("input_cost_per_token", "")),
str(group.get("output_cost_per_token", "")),
_rendered_field(group, "model_group", ""),
_rendered_field(group, "mode", "chat"),
_rendered_field(group, "input_cost_per_token", ""),
_rendered_field(group, "output_cost_per_token", ""),
)
rich.print(table)

3
litellm/proxy/client/cli/commands/up.py
View file

@ -166,7 +166,8 @@ def up(ctx: click.Context) -> None:
is already running (this does not start one for you). Cursor is not
supported: it has no equivalent file-based config to patch.
"""
base_url: Final = ctx.obj["base_url"]
ctx_obj: Final[CliContextObj] = ctx.obj
base_url: Final = ctx_obj["base_url"]
try:
_ensure_fresh_login(ctx)

2
litellm/proxy/common_utils/cache_pydantic_utils.py
View file

@ -37,7 +37,7 @@ class CacheCodec:
"""
@staticmethod
def serialize(value: Any, model_type: type[T] | None = None) -> Any:
def serialize(value: object, model_type: type[T] | None = None) -> object:
"""
Encode a value for DualCache / Redis (``json.dumps``-safe).

4
litellm/proxy/common_utils/proxy_rate_limit_error.py
View file

@ -66,7 +66,7 @@ def map_v3_rate_limit_type(
return None
def _coerce_message(detail: Any) -> str:
def _coerce_message(detail: object) -> str:
"""Best-effort, JSON-friendly stringification of an HTTPException-style detail."""
if detail is None:
return ""
@ -144,7 +144,7 @@ class ProxyRateLimitError(HTTPException, RateLimitError):
def __init__(
self,
detail: Any,
headers: Mapping[str, Any] | None = None,
headers: Mapping[str, object] | None = None,
category: str | RateLimitErrorCategory = RateLimitErrorCategory.LITELLM_RATE_LIMIT,
rate_limit_type: str | RateLimitType | None = None,
model: str | None = None,

10
litellm/proxy/container_endpoints/endpoints.py
View file

@ -106,7 +106,7 @@ async def create_container(
# Process request using ProxyBaseLLMRequestProcessing
processor: Final = ProxyBaseLLMRequestProcessing(data=data)
try:
response: Final = await processor.base_process_llm_request(
response: Final[object] = await processor.base_process_llm_request(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
@ -216,7 +216,7 @@ async def list_containers(
or get_custom_llm_provider_from_request_query(request=request)
or "openai"
)
data: Final[dict[str, Any]] = {
data: Final[dict[str, object]] = {
"query_params": query_params,
"model": query_params.get("model"),
"order": order,
@ -341,7 +341,7 @@ async def retrieve_container(
# Process request using ProxyBaseLLMRequestProcessing
processor: Final = ProxyBaseLLMRequestProcessing(data=data)
try:
return await processor.base_process_llm_request(
container: Final[object] = await processor.base_process_llm_request(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
@ -366,6 +366,7 @@ async def retrieve_container(
proxy_logging_obj=proxy_logging_obj,
version=version,
)
return container
@router.delete(
@ -446,7 +447,7 @@ async def delete_container(
# Process request using ProxyBaseLLMRequestProcessing
processor: Final = ProxyBaseLLMRequestProcessing(data=data)
try:
return await processor.base_process_llm_request(
deleted_container: Final[object] = await processor.base_process_llm_request(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
@ -471,6 +472,7 @@ async def delete_container(
proxy_logging_obj=proxy_logging_obj,
version=version,
)
return deleted_container
# Register JSON-configured container file endpoints

16
litellm/proxy/db/exception_handler.py
View file

@ -1,5 +1,5 @@
from collections.abc import Awaitable, Callable, Iterator
from typing import Any, Final, TypeVar
from typing import Final, Protocol, TypeVar
from litellm._logging import verbose_proxy_logger
from litellm.proxy._types import (
@ -407,8 +407,20 @@ def _coerce_timeout(value: object, fallback: float) -> float:
_ReadResultT: Final = TypeVar("_ReadResultT")
class _DBReconnectClient(Protocol):
"""The one method `call_with_db_reconnect_retry` needs from a Prisma client."""
async def attempt_db_reconnect(
self,
*,
reason: str,
timeout_seconds: float | None = None,
lock_timeout_seconds: float | None = None,
) -> bool: ...
async def call_with_db_reconnect_retry(
prisma_client: Any,
prisma_client: _DBReconnectClient,
coro_factory: Callable[[], Awaitable[_ReadResultT]],
*,
reason: str,

2
litellm/proxy/db/spend_log_tool_index.py
View file

@ -148,4 +148,4 @@ async def flush_tool_usage_transactions(
except DB_RETRY_SAFE_ERROR_TYPES:
if attempt >= n_retry_times:
raise
await asyncio.sleep(2**attempt + random.uniform(0, 1))
await asyncio.sleep(2.0**attempt + random.uniform(0, 1))

16
litellm/proxy/guardrails/guardrail_hooks/crowdstrike_aidr/crowdstrike_aidr.py
View file

@ -6,7 +6,7 @@ from typing import TYPE_CHECKING, Annotated, Final, Literal, NamedTuple, Optiona
from fastapi import HTTPException
from pydantic import BaseModel, ConfigDict, Field, ValidationError
from typing_extensions import Any, override
from typing_extensions import override
from litellm._logging import verbose_proxy_logger
from litellm.integrations.custom_guardrail import (
@ -78,7 +78,7 @@ class _GuardChatCompletionsResult(BaseModel):
"""Whether or not the prompt triggered a block detection."""
transformed: bool | None = None
"""Whether or not the original input was transformed."""
detectors: dict[str, Any] | None = None
detectors: dict[str, object] | None = None
"""Result of the policy analyzing and input prompt."""
@ -146,8 +146,8 @@ def _extract_text_from_message(message: _Message) -> str:
return "\n".join(part.text for part in content if isinstance(part, _TextContentPart))
def _merge_metadata_bags(request_data: Mapping[str, Any]) -> Mapping[str, Any] | None:
merged: Final[dict[str, Any]] = {}
def _merge_metadata_bags(request_data: Mapping[str, object]) -> Mapping[str, object] | None:
merged: Final[dict[str, object]] = {}
present = False
for bag in (request_data.get("metadata"), request_data.get("litellm_metadata")):
if isinstance(bag, Mapping):
@ -313,7 +313,7 @@ class CrowdStrikeAIDRHandler(CustomGuardrail):
self._set_streaming_params(streaming_params_from_litellm_params(litellm_params))
async def _call_crowdstrike_aidr_guard(
self, payload: dict[str, Any], hook_name: str
self, payload: dict[str, object], hook_name: str
) -> _GuardChatCompletionsResult:
"""
Makes the API call to the CrowdStrike AIDR AI Guard endpoint.
@ -423,7 +423,7 @@ class CrowdStrikeAIDRHandler(CustomGuardrail):
return [_extract_text_from_message(msg) for msg in tail]
async def _call_or_fail_open(
self, payload: dict[str, Any], hook_name: str, request_data: dict[str, object]
self, payload: dict[str, object], hook_name: str, request_data: dict[str, object]
) -> _GuardChatCompletionsResult:
start_time: Final = time.time()
try:
@ -506,7 +506,7 @@ class CrowdStrikeAIDRHandler(CustomGuardrail):
event_type = "output"
hook_name = "apply_guardrail (response)"
ai_guard_payload: Final[dict[str, Any]] = {
ai_guard_payload: Final[dict[str, object]] = {
"guard_input": guard_input.model_dump(mode="json"),
"event_type": event_type,
}
@ -521,7 +521,7 @@ class CrowdStrikeAIDRHandler(CustomGuardrail):
if user_id:
ai_guard_payload["user_id"] = user_id
extra_info: Final[dict[str, str]] = {}
extra_info: Final[dict[str, object]] = {}
user_email: Final = metadata.get("user_api_key_user_email")
if user_email:
extra_info["user_name"] = user_email

4
litellm/proxy/guardrails/guardrail_hooks/enkryptai/enkryptai.py
View file

@ -6,7 +6,7 @@
# +-------------------------------------------------------------+
import os
from collections.abc import AsyncGenerator
from collections.abc import AsyncGenerator, AsyncIterable
from datetime import datetime
from typing import TYPE_CHECKING, Any, Final, Literal, Optional
@ -465,7 +465,7 @@ class EnkryptAIGuardrails(CustomGuardrail):
async def async_post_call_streaming_iterator_hook(
self,
user_api_key_dict: UserAPIKeyAuth,
response: Any,
response: AsyncIterable[ModelResponseStream],
request_data: dict,
) -> AsyncGenerator[ModelResponseStream, None]:
"""

2
litellm/proxy/guardrails/guardrail_hooks/qualifire/qualifire.py
View file

@ -383,7 +383,7 @@ class QualifireGuardrail(CustomGuardrail):
result: Final = response.json()
# Extract response info for logging
qualifire_response: Final = {
qualifire_response: Final[dict[str, object]] = {
"score": result.get("score"),
"status": result.get("status"),
}

3
litellm/proxy/guardrails/guardrail_hooks/semantic_guard/route_loader.py
View file

@ -6,6 +6,7 @@ then builds a SemanticRouter for prompt matching.
"""
import os
from collections.abc import Mapping, Sequence
from typing import TYPE_CHECKING, Any, Final
import yaml
@ -66,7 +67,7 @@ class SemanticGuardRouteLoader:
cls,
route_templates: list[str] | None,
custom_routes_file: str | None,
custom_routes: list[dict[str, Any]] | None,
custom_routes: Sequence[Mapping[str, object]] | None,
global_threshold: float = DEFAULT_SEMANTIC_GUARD_SIMILARITY_THRESHOLD,
) -> list["Route"]:
"""Build semantic-router Route objects from templates + custom config."""

6
litellm/proxy/hooks/parallel_request_limiter.py
View file

@ -27,7 +27,7 @@ if TYPE_CHECKING:
from litellm.proxy.utils import InternalUsageCache as _InternalUsageCache
Span = _Span | Any
Span = _Span
InternalUsageCache = _InternalUsageCache
else:
Span = Any
@ -75,7 +75,7 @@ class _PROXY_MaxParallelRequestsHandler(CustomLogger):
current: dict | None,
request_count_api_key: str,
rate_limit_type: Literal["key", "model_per_key", "user", "customer", "team"],
values_to_update_in_cache: list[tuple[Any, Any]],
values_to_update_in_cache: list[tuple[str, object]],
) -> dict:
verbose_proxy_logger.info("Current Usage of %s in this minute: %s", rate_limit_type, current)
if current is None:
@ -266,7 +266,7 @@ class _PROXY_MaxParallelRequestsHandler(CustomLogger):
rpm_limit = sys.maxsize
values_to_update_in_cache: list[
tuple[Any, Any]
tuple[str, object]
] = [] # values that need to get updated in cache, will run a batch_set_cache after this function
# ------------

7
litellm/proxy/management_endpoints/common_daily_activity.py
View file

@ -17,6 +17,7 @@ from litellm.proxy.spend_tracking.key_metadata_recovery import (
)
from litellm.proxy.spend_tracking.ptu_feature_flag import is_ptu_cost_attribution_enabled
from litellm.proxy.utils import PrismaClient
from litellm.repositories.prisma_protocols import TableActions
from litellm.repositories.table_repositories import DeletedVerificationTokenRepository
from litellm.repositories.verification_token_repository import (
VerificationTokenRepository,
@ -1155,8 +1156,10 @@ async def get_daily_activity(
include_current_utc_day=include_current_utc_day,
)
spend_table: Final[TableActions[DailySpendRecord]] = getattr(prisma_client.db, table_name)
# Get total count for pagination
total_count: Final[int] = await getattr(prisma_client.db, table_name).count(where=where_conditions)
total_count: Final[int] = await spend_table.count(where=where_conditions)
# Fetch paginated results.
# ``date`` alone is not a unique sort key -- a busy tenant has many
@ -1168,7 +1171,7 @@ async def get_daily_activity(
# total. Adding ``id`` (the row's UUID primary key, present on both
# LiteLLM_DailyUserSpend and LiteLLM_DailyTeamSpend) as a tiebreaker
# gives every page a stable cursor (#30164).
daily_spend_data: Final[Sequence[DailySpendRecord]] = await getattr(prisma_client.db, table_name).find_many(
daily_spend_data: Final[Sequence[DailySpendRecord]] = await spend_table.find_many(
where=where_conditions,
order=[
{"date": "desc"},

14
litellm/proxy/ocr_endpoints/endpoints.py
View file

@ -2,7 +2,7 @@
import json
from collections.abc import Mapping
from typing import Any, Final, cast
from typing import Final, cast
import orjson
from fastapi import APIRouter, Depends, HTTPException, Request, Response, UploadFile
@ -48,7 +48,7 @@ def _build_document_from_upload(
)
def _with_request_format(data: Mapping[str, Any], request: Request) -> Mapping[str, Any]:
def _with_request_format(data: Mapping[str, object], request: Request) -> Mapping[str, object]:
"""
Resolve the requested response format from the body or the `x-req-format` header.
@ -90,7 +90,7 @@ def _native_response(response: object, fastapi_response: Response) -> Response |
)
async def _parse_multipart_form(request: Request) -> dict[str, Any]:
async def _parse_multipart_form(request: Request) -> dict[str, object]:
"""
Extract OCR data from a multipart form request.
@ -130,7 +130,7 @@ async def _parse_multipart_form(request: Request) -> dict[str, Any]:
content_type=uploaded_file.content_type,
)
data: Final[dict[str, Any]] = {"document": document}
data: Final[dict[str, object]] = {"document": document}
for field_name, field_value in form.items():
if field_name in ("file", "document"):
@ -154,12 +154,12 @@ async def _parse_multipart_form(request: Request) -> dict[str, Any]:
return data
async def _parse_ocr_request(request: Request) -> Mapping[str, Any]:
async def _parse_ocr_request(request: Request) -> Mapping[str, object]:
"""Parse an OCR request and apply the `x-req-format` header, if any."""
return _with_request_format(await _parse_ocr_request_body(request), request)
async def _parse_ocr_request_body(request: Request) -> dict[str, Any]:
async def _parse_ocr_request_body(request: Request) -> dict[str, object]:
"""
Parse an OCR request, supporting both JSON and multipart form data.
@ -320,7 +320,7 @@ async def ocr(
# Process request using ProxyBaseLLMRequestProcessing
processor = ProxyBaseLLMRequestProcessing(data=data)
response: Final = await processor.base_process_llm_request(
response: Final[object] = await processor.base_process_llm_request(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,

28
litellm/repositories/budget_repository.py
View file

@ -2,7 +2,8 @@
Budget repository for database operations on LiteLLM_BudgetTable.
"""
from typing import TYPE_CHECKING, Any, Final
from collections.abc import Mapping
from typing import TYPE_CHECKING, Final, Protocol
from litellm.models.budget import LiteLLM_BudgetTable
from litellm.repositories.base_repository import BaseRepository
@ -12,12 +13,27 @@ if TYPE_CHECKING:
from prisma import models as prisma_models
class _BudgetDb(Protocol):
"""The single Prisma table this repository reaches for on ``prisma_client.db``."""
@property
def litellm_budgettable(self) -> TableActions["prisma_models.LiteLLM_BudgetTable"]: ...
class _PrismaClientView(Protocol):
"""The one attribute this repository reads off the untyped Prisma client wrapper."""
@property
def db(self) -> _BudgetDb: ...
class BudgetRepository(BaseRepository[LiteLLM_BudgetTable]):
"""Repository for budget database operations."""
@property
def table(self) -> TableActions["prisma_models.LiteLLM_BudgetTable"]:
return self.prisma_client.db.litellm_budgettable
client: Final[_PrismaClientView] = self.prisma_client
return client.db.litellm_budgettable
@property
def model_class(self) -> type[LiteLLM_BudgetTable]:
@ -34,12 +50,12 @@ class BudgetRepository(BaseRepository[LiteLLM_BudgetTable]):
max_parallel_requests: int | None = None,
tpm_limit: int | None = None,
rpm_limit: int | None = None,
model_max_budget: dict[str, Any] | None = None,
model_max_budget: Mapping[str, object] | None = None,
budget_duration: str | None = None,
allowed_models: list[str] | None = None,
) -> LiteLLM_BudgetTable:
"""Create a new budget record."""
data: Final[dict[str, Any]] = {
data: Final[dict[str, object]] = {
"created_by": created_by,
"updated_by": created_by,
}
@ -71,12 +87,12 @@ class BudgetRepository(BaseRepository[LiteLLM_BudgetTable]):
max_parallel_requests: int | None = None,
tpm_limit: int | None = None,
rpm_limit: int | None = None,
model_max_budget: dict[str, Any] | None = None,
model_max_budget: Mapping[str, object] | None = None,
budget_duration: str | None = None,
allowed_models: list[str] | None = None,
) -> LiteLLM_BudgetTable | None:
"""Update an existing budget record."""
data: Final[dict[str, Any]] = {"updated_by": updated_by}
data: Final[dict[str, object]] = {"updated_by": updated_by}
if max_budget is not None:
data["max_budget"] = max_budget
if soft_budget is not None:

28
litellm/repositories/organization_repository.py
View file

@ -2,7 +2,8 @@
Organization repository for database operations on LiteLLM_OrganizationTable.
"""
from typing import TYPE_CHECKING, Any, Final
from collections.abc import Mapping
from typing import TYPE_CHECKING, Final, Protocol
from litellm.models.organization import LiteLLM_OrganizationTable
from litellm.repositories.base_repository import BaseRepository
@ -12,12 +13,27 @@ if TYPE_CHECKING:
from prisma import models as prisma_models
class _OrganizationDb(Protocol):
"""The single Prisma table this repository reaches for on ``prisma_client.db``."""
@property
def litellm_organizationtable(self) -> TableActions["prisma_models.LiteLLM_OrganizationTable"]: ...
class _PrismaClientView(Protocol):
"""The one attribute this repository reads off the untyped Prisma client wrapper."""
@property
def db(self) -> _OrganizationDb: ...
class OrganizationRepository(BaseRepository[LiteLLM_OrganizationTable]):
"""Repository for organization database operations."""
@property
def table(self) -> TableActions["prisma_models.LiteLLM_OrganizationTable"]:
return self.prisma_client.db.litellm_organizationtable
client: Final[_PrismaClientView] = self.prisma_client
return client.db.litellm_organizationtable
@property
def model_class(self) -> type[LiteLLM_OrganizationTable]:
@ -39,12 +55,12 @@ class OrganizationRepository(BaseRepository[LiteLLM_OrganizationTable]):
budget_id: str,
created_by: str,
organization_id: str | None = None,
metadata: dict[str, Any] | None = None,
metadata: Mapping[str, object] | None = None,
models: list[str] | None = None,
object_permission_id: str | None = None,
) -> LiteLLM_OrganizationTable:
"""Create a new organization."""
data: Final[dict[str, Any]] = {
data: Final[dict[str, object]] = {
"organization_alias": organization_alias,
"budget_id": budget_id,
"created_by": created_by,
@ -67,12 +83,12 @@ class OrganizationRepository(BaseRepository[LiteLLM_OrganizationTable]):
updated_by: str,
organization_alias: str | None = None,
budget_id: str | None = None,
metadata: dict[str, Any] | None = None,
metadata: Mapping[str, object] | None = None,
models: list[str] | None = None,
object_permission_id: str | None = None,
) -> LiteLLM_OrganizationTable | None:
"""Update an organization."""
data: Final[dict[str, Any]] = {"updated_by": updated_by}
data: Final[dict[str, object]] = {"updated_by": updated_by}
if organization_alias is not None:
data["organization_alias"] = organization_alias
if budget_id is not None:

11
litellm/repositories/project_repository.py
View file

@ -2,7 +2,8 @@
Project repository for database operations on LiteLLM_ProjectTable.
"""
from typing import TYPE_CHECKING, Any, Final
from collections.abc import Mapping
from typing import TYPE_CHECKING, Final
from litellm.models.project import LiteLLM_ProjectTable
from litellm.repositories.base_repository import BaseRepository
@ -43,14 +44,14 @@ class ProjectRepository(BaseRepository[LiteLLM_ProjectTable]):
description: str | None = None,
team_id: str | None = None,
budget_id: str | None = None,
metadata: dict[str, Any] | None = None,
metadata: Mapping[str, object] | None = None,
models: list[str] | None = None,
model_rpm_limit: dict[str, int] | None = None,
model_tpm_limit: dict[str, int] | None = None,
object_permission_id: str | None = None,
) -> LiteLLM_ProjectTable:
"""Create a new project."""
data: Final[dict[str, Any]] = {
data: Final[dict[str, object]] = {
"created_by": created_by,
"updated_by": created_by,
}
@ -85,7 +86,7 @@ class ProjectRepository(BaseRepository[LiteLLM_ProjectTable]):
description: str | None = None,
team_id: str | None = None,
budget_id: str | None = None,
metadata: dict[str, Any] | None = None,
metadata: Mapping[str, object] | None = None,
models: list[str] | None = None,
model_rpm_limit: dict[str, int] | None = None,
model_tpm_limit: dict[str, int] | None = None,
@ -93,7 +94,7 @@ class ProjectRepository(BaseRepository[LiteLLM_ProjectTable]):
object_permission_id: str | None = None,
) -> LiteLLM_ProjectTable | None:
"""Update a project."""
data: Final[dict[str, Any]] = {"updated_by": updated_by}
data: Final[dict[str, object]] = {"updated_by": updated_by}
if project_alias is not None:
data["project_alias"] = project_alias
if description is not None:

9
litellm/router_strategy/adaptive_router/adaptive_router.py
View file

@ -16,6 +16,7 @@ from __future__ import annotations
import asyncio
import time
from collections import OrderedDict
from collections.abc import Mapping
from dataclasses import asdict, dataclass
from typing import Any, Final, cast
@ -122,7 +123,7 @@ class AdaptiveRouter:
prefs = self.model_to_prefs.get(model) or _default_prefs()
self._cells[(rt, model)] = initial_cell(prefs, rt)
async def load_state_from_db(self, prisma_client: Any) -> None:
async def load_state_from_db(self, prisma_client: object) -> None:
"""Add each row's persisted delta to a freshly computed cold-start prior.
A row holds an accumulated delta, not a full posterior, and can be one-sided
@ -237,7 +238,7 @@ class AdaptiveRouter:
cost_weight=self.config.weights.cost,
)
async def get_state_snapshot(self) -> dict[str, Any]:
async def get_state_snapshot(self) -> dict[str, object]:
"""In-memory snapshot for the introspection endpoint. Cheap; no DB hit."""
cells: Final = []
for (rt, model), cell in sorted(self._cells.items(), key=lambda kv: (kv[0][0].value, kv[0][1])):
@ -278,7 +279,7 @@ class AdaptiveRouter:
@staticmethod
def _extract_min_quality_tier(
request_kwargs: dict[str, Any],
request_kwargs: Mapping[str, object],
) -> int | None:
"""Pull `min_quality_tier` from request headers or metadata.
@ -484,7 +485,7 @@ class AdaptiveRouter:
return combined_delta
@staticmethod
def _persistable_session_snapshot(state: SessionState) -> dict[str, Any]:
def _persistable_session_snapshot(state: SessionState) -> dict[str, object]:
snapshot: Final = asdict(state)
for sensitive in (
"last_user_content",

8
litellm/router_strategy/adaptive_router/signals.py
View file

@ -92,7 +92,7 @@ class Turn:
user_content: str | None = None
assistant_content: str | None = None
tool_calls: list[dict[str, Any]] = field(default_factory=list)
tool_calls: Sequence[Mapping[str, object]] = field(default_factory=list[Mapping[str, object]])
tool_results: Sequence[Mapping[str, object]] = field(default_factory=list)
response_status: int | None = None
@ -174,7 +174,7 @@ def _detect_failure(tool_results: Sequence[Mapping[str, object]]) -> bool:
return False
def _signature(call: dict[str, Any]) -> str:
def _signature(call: Mapping[str, Any]) -> str:
"""Stable signature for loop detection: name + sorted JSON-ish args."""
name: Final = call.get("name") or call.get("function", {}).get("name", "")
call_args = call.get("arguments")
@ -185,7 +185,7 @@ def _signature(call: dict[str, Any]) -> str:
return f"{name}({call_args})"
def _detect_loop(history: list[str], new_calls: list[dict[str, Any]]) -> bool:
def _detect_loop(history: list[str], new_calls: Sequence[Mapping[str, object]]) -> bool:
"""Fires if any new call's signature appears >= LOOP_REPEAT_THRESHOLD-1 times
in recent history (so this call would be the Nth)."""
if not new_calls:
@ -238,7 +238,7 @@ def detect_response_signals(
previous_assistant_content: str | None,
current_assistant_content: str | None,
tool_call_history: list[str],
tool_calls: list[dict[str, Any]],
tool_calls: Sequence[Mapping[str, object]],
tool_results: Sequence[Mapping[str, object]],
response_status: int | None,
) -> SignalDelta:

11
litellm/router_strategy/adaptive_router/update_queue.py
View file

@ -19,7 +19,8 @@ to the in-memory aggregator). Flush is async and batched.
from __future__ import annotations
import asyncio
from typing import Any, Final
from collections.abc import Mapping
from typing import Final
from litellm._logging import verbose_router_logger
from litellm.repositories.table_repositories import (
@ -39,7 +40,7 @@ class AdaptiveRouterUpdateQueue:
def __init__(self) -> None:
self._state_agg: dict[StateKey, dict[str, float]] = {}
self._session_agg: dict[SessionKey, dict[str, Any]] = {}
self._session_agg: dict[SessionKey, Mapping[str, object]] = {}
self._lock = asyncio.Lock()
self._max_state_size_seen = 0
self._max_session_size_seen = 0
@ -77,7 +78,7 @@ class AdaptiveRouterUpdateQueue:
session_id: str,
router_name: str,
model_name: str,
state_dict: dict[str, Any],
state_dict: Mapping[str, object],
) -> None:
"""
Last-write-wins per session row. The state_dict is a snapshot of the
@ -91,7 +92,7 @@ class AdaptiveRouterUpdateQueue:
# ---- Flushers (called by background task) ----------------------------
async def flush_state_to_db(self, prisma_client: Any) -> int:
async def flush_state_to_db(self, prisma_client: object) -> int:
"""
Drain state aggregator and apply to LiteLLM_AdaptiveRouterState.
Returns number of cells flushed.
@ -147,7 +148,7 @@ class AdaptiveRouterUpdateQueue:
return len(batch)
async def flush_session_to_db(self, prisma_client: Any) -> int:
async def flush_session_to_db(self, prisma_client: object) -> int:
"""
Drain session aggregator and upsert into LiteLLM_AdaptiveRouterSession.
Returns number of session rows flushed.

2
litellm/types/containers/main.py
View file

@ -140,7 +140,7 @@ class ContainerFileObject(BaseModel):
created_at: int
path: str
source: str
_hidden_params: dict[str, Any] = {}
_hidden_params: dict[str, builtins.object] = {}
def __contains__(self, key: str) -> bool:
return hasattr(self, key)

16
litellm/types/guardrails.py
View file

@ -2,10 +2,10 @@ from collections.abc import Mapping
from datetime import datetime
from enum import Enum
from types import MappingProxyType
from typing import Any, Final, Literal
from typing import Final, Literal
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from typing_extensions import Required, TypedDict
from typing_extensions import ReadOnly, Required, TypedDict
from litellm.constants import BEDROCK_APPLY_GUARDRAIL_CHUNK_BUDGET_CHARS
from litellm.types.proxy.guardrails.guardrail_hooks.akto import (
@ -935,7 +935,7 @@ class BaseLitellmParams(ContentFilterConfigModel): # works for new and patch up
),
)
additional_provider_specific_params: dict[str, Any] | None = Field(
additional_provider_specific_params: dict[str, object] | None = Field(
default=None,
description="Additional provider-specific parameters for generic guardrail APIs",
)
@ -1157,7 +1157,7 @@ class GuardrailEventHooks(str, Enum):
class DynamicGuardrailParams(TypedDict):
extra_body: dict[str, Any]
extra_body: ReadOnly[dict[str, object]]
class GUARDRAIL_DEFINITION_LOCATION(str, Enum):
@ -1188,7 +1188,7 @@ class GuardrailUIAddGuardrailSettings(BaseModel):
supported_modes: list[str]
supported_modes_by_provider: dict[str, list[str]]
pii_entity_categories: list[PiiEntityCategoryMap]
content_filter_settings: dict[str, Any] | None = None
content_filter_settings: dict[str, object] | None = None
class PresidioPerRequestConfig(BaseModel):
@ -1206,8 +1206,8 @@ class ApplyGuardrailRequest(BaseModel):
language: str | None = None
entities: list[PiiEntityType] | None = None
input_type: str = "request"
messages: list[dict[str, Any]] | None = None
metadata: dict[str, Any] | None = None
messages: list[dict[str, object]] | None = None
metadata: dict[str, object] | None = None
class ApplyGuardrailResponse(BaseModel):
@ -1217,4 +1217,4 @@ class ApplyGuardrailResponse(BaseModel):
class PatchGuardrailRequest(BaseModel):
guardrail_name: str | None = None
litellm_params: BaseLitellmParams | None = None
guardrail_info: dict[str, Any] | None = None
guardrail_info: dict[str, object] | None = None

8
litellm/types/integrations/prometheus.py
View file

@ -44,7 +44,7 @@ def _sanitize_prometheus_label_name(label: str) -> str:
_PROMETHEUS_LABEL_VALUE_TRANSLATE_V1: Final = str.maketrans("\n", " ", "\r\u2028\u2029")
def _sanitize_prometheus_label_value(value: Any | None) -> str | None:
def _sanitize_prometheus_label_value(value: object | None) -> str | None:
"""
Same semantics as :func:`_sanitize_prometheus_label_value`, implemented with
``str.translate`` plus a single escape pass instead of chained ``replace``.
@ -1023,7 +1023,7 @@ class UserAPIKeyLabelValues:
``hashed_api_key``. This supports ``**standard_logging_payload`` in tests.
"""
field_names: Final = {f.name for f in fields(self)}
merged: Final[dict[str, Any]] = {}
merged: Final[dict[str, object]] = {}
for f in fields(self):
if f.default_factory is not MISSING:
merged[f.name] = f.default_factory()
@ -1060,9 +1060,9 @@ class UserAPIKeyLabelValues:
# stays cheap. (Dataclass default `str()` delegates to `__repr__`.)
return ""
def model_dump(self) -> dict[str, Any]:
def model_dump(self) -> dict[str, object]:
"""Same shape as the former Pydantic ``model_dump()`` (plain dict, list tags)."""
d: Final[dict[str, Any]] = {f.name: getattr(self, f.name) for f in fields(self)}
d: Final[dict[str, object]] = {f.name: getattr(self, f.name) for f in fields(self)}
d["tags"] = list(self.tags)
d["custom_metadata_labels"] = dict(self.custom_metadata_labels)
return d

16
litellm/types/realtime.py
View file

@ -78,15 +78,15 @@ class RealtimeSessionConfig(BaseModel):
type: str | None = None
model: str | None = None
instructions: str | None = None
audio: dict[str, Any] | None = None
audio: dict[str, object] | None = None
include: list[str] | None = None
max_output_tokens: int | str | None = None
output_modalities: list[str] | None = None
tool_choice: Any | None = None
tools: list[dict[str, Any]] | None = None
tracing: Any | None = None
truncation: Any | None = None
prompt: dict[str, Any] | None = None
tool_choice: object | None = None
tools: list[dict[str, object]] | None = None
tracing: object | None = None
truncation: object | None = None
prompt: dict[str, object] | None = None
class RealtimeClientSecretRequest(BaseModel):
@ -114,7 +114,7 @@ class RealtimeClientSecretResponse(BaseModel):
expires_at: int | None = None
value: str
session: dict[str, Any] | None = None
session: dict[str, object] | None = None
class RealtimeTranscriptionSessionRequest(BaseModel):
@ -151,7 +151,7 @@ class RealtimeTranscriptionSessionResponse(BaseModel):
model_config = {"extra": "allow"}
client_secret: dict[str, Any] | None = None
client_secret: dict[str, object] | None = None
class RealtimeErrorDetail(TypedDict):

11
litellm/vector_store_files/utils.py
View file

@ -1,4 +1,5 @@
from typing import Any, Final, cast, get_type_hints
from collections.abc import Mapping
from typing import Final, cast, get_type_hints
from litellm.types.vector_store_files import (
VectorStoreFileCreateRequest,
@ -11,25 +12,25 @@ class VectorStoreFileRequestUtils:
"""Helper utilities for constructing vector store file requests."""
@staticmethod
def _filter_params(params: dict[str, Any], model: Any) -> dict[str, Any]:
def _filter_params(params: Mapping[str, object], model: type[object]) -> dict[str, object]:
valid_keys: Final = get_type_hints(model).keys()
return {key: value for key, value in params.items() if key in valid_keys and value is not None}
@staticmethod
def get_create_request_params(
params: dict[str, Any],
params: Mapping[str, object],
) -> VectorStoreFileCreateRequest:
filtered: Final = VectorStoreFileRequestUtils._filter_params(params=params, model=VectorStoreFileCreateRequest)
return cast(VectorStoreFileCreateRequest, filtered)
@staticmethod
def get_list_query_params(params: dict[str, Any]) -> VectorStoreFileListQueryParams:
def get_list_query_params(params: Mapping[str, object]) -> VectorStoreFileListQueryParams:
filtered = VectorStoreFileRequestUtils._filter_params(params=params, model=VectorStoreFileListQueryParams)
return cast(VectorStoreFileListQueryParams, filtered)
@staticmethod
def get_update_request_params(
params: dict[str, Any],
params: Mapping[str, object],
) -> VectorStoreFileUpdateRequest:
filtered: Final = VectorStoreFileRequestUtils._filter_params(params=params, model=VectorStoreFileUpdateRequest)
return cast(VectorStoreFileUpdateRequest, filtered)