scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-26 01:12:21 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 98

refactor: daily fresh tech debt cleanup, rolling PR (#42710)

Browse source 
* refactor: clear fresh tech debt from the last 24 hours (2026-09-05, 2026-09-06)

Drop the TID251 cast import and both cast-ok casts from the refusal
message_delta rebuild by narrowing the TypedDict union on its type literal,
drop the redundant Mapping cast after the isinstance check in
_mapping_field, and type the Lyria predict read-only helpers as
Mapping[str, object] instead of a bare dict with mutable-ok.

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* refactor: clear fresh tech debt from the last 24 hours (2026-09-09)

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* refactor: clear fresh tech debt from the last 24 hours (2026-09-10)

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* refactor: keep the pre-existing cost-estimate comment and usage cost read out of the cleanup

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* refactor: clear fresh tech debt from the last 24 hours (2026-09-13)

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* refactor: keep the model info pricing helper out of the cleanup

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* refactor: drop suppressions that no longer suppress anything (2026-09-16)

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* refactor: keep the rebind-ok reason inside the line limit

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(anthropic): rebuild the refusal message_delta by spreading the chunk

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* refactor: clear fresh tech debt from the last 24 hours (2026-09-17)

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* refactor: clear fresh tech debt from the last 24 hours (2026-09-18)

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* refactor: clear fresh tech debt from the last 24 hours (2026-09-19)

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* refactor: keep the pre-existing protected-resource return type out of the cleanup

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* refactor: clear fresh tech debt from the last 24 hours (2026-09-20)

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* refactor: type fresh getattr, Any, and bare dict debt from 2026-09-22

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(mcp): keep the string guard on tools/list next_cursor

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* refactor(vercel_ai_gateway): type the embedding error headers dict

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* chore(techdebt): fix inert suppressions and missing Final in 2026-09-22 changes

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* chore(techdebt): shorten suppression reason to fit line length

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* chore(techdebt): format provider spread so its suppression sits on the literal

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* chore(techdebt): drop the logger extras suppression that LIT013 now flags as inert

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* chore(techdebt): clear fresh suppressions, Any aliases and slop from the 24h window

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* refactor(vercel): take a read-only headers mapping in get_error_class

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: mateo <mateo@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-09-24 04:17:23 -07:00 • committed by GitHub
parent 8bbe7edb71
commit d17c0d7724
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
45 changed files with 96 additions and 107 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
litellm/cost_calculator.py
View file

@ -2017,8 +2017,8 @@ def _deployment_model_info(
return cast(ModelInfo, registered_deployment_info) # cast-ok: router registers deployment prices under its id
if litellm_logging_obj is None:
return None
litellm_params: Final = getattr(litellm_logging_obj, "litellm_params", None)
if litellm_params is None:
litellm_params: Final = litellm_logging_obj.litellm_params
if not litellm_params:
return None
return next(
(
@ -2036,7 +2036,9 @@ def _ocr_model_info(
router_model_id: str | None,
) -> OCRPricing | None:
deployment_info: Final = _deployment_model_info(litellm_logging_obj, custom_pricing, router_model_id)
litellm_params: Final = getattr(litellm_logging_obj, "litellm_params", None) if custom_pricing else None
litellm_params: Final = (
litellm_logging_obj.litellm_params if custom_pricing and litellm_logging_obj is not None else None
)
if litellm_params is None:
return deployment_info
return _layered_ocr_pricing(litellm_params, deployment_info)

2
litellm/experimental_mcp_client/tools.py
View file

@ -129,7 +129,7 @@ async def list_tools_with_pagination(
)
tools.extend(result.tools)
next_cursor = getattr(result, "next_cursor", None)
next_cursor = result.next_cursor
if not isinstance(next_cursor, str) or not next_cursor:
return tools
if next_cursor in seen_cursors:

2
litellm/integrations/arize/arize.py
View file

@ -112,7 +112,7 @@ class ArizeLogger(OpenTelemetry):
if value is None or value in ("", "None"):
return None
try:
rate = float(value)
rate: Final = float(value)
except (TypeError, ValueError):
verbose_logger.warning(
"ArizeLogger: %s value %r is not a number; exporting the request",

24
litellm/integrations/mavvrik_focus/mavvrik_focus_logger.py
View file

@ -21,7 +21,7 @@ from __future__ import annotations
import os
from datetime import datetime, timedelta, timezone
from typing import TYPE_CHECKING, Any, Final, Protocol
from typing import TYPE_CHECKING, Any, Final
import litellm
from litellm._logging import verbose_proxy_logger
@ -35,17 +35,6 @@ else:
AsyncIOScheduler = Any
class _PodLockManager(Protocol):
"""The subset of PodLockManager this logger drives to serialize the export across pods."""
@property
def redis_cache(self) -> object: ...
async def acquire_lock(self, cronjob_id: str) -> bool | None: ...
async def release_lock(self, cronjob_id: str) -> None: ...
def _parse_metrics_marker(
marker: object | None,
) -> datetime | None:
@ -237,13 +226,10 @@ class MavvrikFocusLogger(FocusLogger):
"""Scheduler entry point — uses Mavvrik-specific pod-lock key."""
from litellm.proxy.proxy_server import proxy_logging_obj # noqa: PLC0415
pod_lock_manager: _PodLockManager | None = None
if proxy_logging_obj is not None:
writer: Final[object] = getattr(proxy_logging_obj, "db_spend_update_writer", None)
if writer is not None:
pod_lock_manager = getattr(writer, "pod_lock_manager", None)
if pod_lock_manager and pod_lock_manager.redis_cache:
pod_lock_manager: Final = (
proxy_logging_obj.db_spend_update_writer.pod_lock_manager if proxy_logging_obj is not None else None
)
if pod_lock_manager is not None and pod_lock_manager.redis_cache:
acquired: Final = await pod_lock_manager.acquire_lock(cronjob_id=MAVVRIK_FOCUS_EXPORT_JOB_NAME)
if not acquired:
verbose_proxy_logger.debug("Mavvrik FOCUS export: unable to acquire pod lock")

2
litellm/integrations/websearch_interception/handler.py
View file

@ -1849,7 +1849,7 @@ class WebSearchInterceptionLogger(CustomLogger):
for tool_call in tool_calls:
# Handle both Anthropic-style input and OpenAI-style function.arguments
query = None
tool_args: dict | None = None # mutable-ok: the tool call's own arguments dict
tool_args: dict[str, object] | None = None # mutable-ok: the tool call's own arguments dict
if "input" in tool_call and isinstance(tool_call["input"], dict):
tool_args = tool_call["input"]
query = tool_args.get("query")

2
litellm/litellm_core_utils/core_helpers.py
View file

@ -365,7 +365,7 @@ def _budget_reservation_on_auth_object(user_api_key_auth: object) -> object:
return getattr(user_api_key_auth, "budget_reservation", None)
def budget_reservation_from_metadata(metadata: Mapping[str, object]) -> dict | None:
def budget_reservation_from_metadata(metadata: Mapping[str, object]) -> dict[str, object] | None:
stamped: Final = metadata.get("user_api_key_budget_reservation")
if isinstance(stamped, dict):
return stamped

4
litellm/litellm_core_utils/litellm_logging.py
View file

@ -5191,7 +5191,7 @@ def _maybe_construct_otel_v2(callback_name: str, _in_memory_loggers: list[Custom
for callback in _in_memory_loggers:
if (
isinstance(callback, OpenTelemetryV2)
and getattr(callback, "callback_name", None) == callback_name
and callback.callback_name == callback_name
and (serves_a_destination or not _exports_nowhere(callback.config))
):
return callback
@ -6663,7 +6663,7 @@ def get_standard_logging_object_payload(
cost_breakdown=request_cost_breakdown,
autorouter_savings=autorouter_savings,
autorouter_savings_estimate=(
{
{ # mutable-ok: spend-log JSON serialization requires plain mappings
"version": 3,
"status": "unknown",
"reason": "pending_projection",

11
litellm/litellm_core_utils/llm_cost_calc/zero_cost_diagnostic.py
View file

@ -5,7 +5,12 @@ from typing import Final
from pydantic import TypeAdapter, ValidationError
from typing_extensions import assert_never
from litellm.types.utils import StandardLoggingZeroCostDiagnostic, Usage
from litellm.types.utils import (
CompletionTokensDetailsWrapper,
PromptTokensDetailsWrapper,
StandardLoggingZeroCostDiagnostic,
Usage,
)
ZERO_COST_COUNTER_NAME: Final = "litellm_zero_cost_requests_total"
@ -18,8 +23,8 @@ _NESTED_PRICING: Final = TypeAdapter(Mapping[str, object] | tuple[object, ...])
_MAX_PRICING_DEPTH: Final = 4
def _audio_tokens(details: object) -> int:
audio_tokens: Final = getattr(details, "audio_tokens", None)
def _audio_tokens(details: PromptTokensDetailsWrapper | CompletionTokensDetailsWrapper | None) -> int:
audio_tokens: Final = details.audio_tokens if details is not None else None
return audio_tokens if isinstance(audio_tokens, int) and audio_tokens > 0 else 0

4
litellm/litellm_core_utils/prompt_templates/common_utils.py
View file

@ -2003,11 +2003,11 @@ def strip_encrypted_reasoning_from_messages(messages: object) -> None:
"""
if not isinstance(messages, list):
return
for content in _anthropic_content_lists(cast(list[object], messages)): # cast-ok: untyped client json
for content in anthropic_content_lists(cast(list[object], messages)): # cast-ok: untyped client json
_strip_encrypted_reasoning_from_blocks(content)
def _anthropic_content_lists(messages: Sequence[object]) -> Iterator[object]:
def anthropic_content_lists(messages: Sequence[object]) -> Iterator[object]:
return (
cast(list[object], content) # cast-ok: narrowed by isinstance
for message in messages

2
litellm/litellm_core_utils/streaming_handler.py
View file

@ -1329,7 +1329,7 @@ class CustomStreamWrapper:
"is_finished": chunk_finish_reason is not None,
"finish_reason": chunk_finish_reason,
"original_chunk": cached_chunk,
"tool_calls": (getattr(cached_choice.delta, "tool_calls", None) if cached_choice is not None else None),
"tool_calls": cached_choice.delta.tool_calls if cached_choice is not None else None,
}
completion_obj["content"] = response_obj["text"]

2
litellm/llms/a2a/chat/transformation.py
View file

@ -48,7 +48,7 @@ def _registry_api_key(agent_litellm_params: Mapping[str, object]) -> str | None:
return configured_api_key if isinstance(configured_api_key, str) else None
def _registry_headers(agent_litellm_params: Mapping[str, object]) -> dict[str, Any] | None:
def _registry_headers(agent_litellm_params: Mapping[str, object]) -> dict[str, object] | None:
stored_headers: Final = agent_litellm_params.get("headers")
if not isinstance(stored_headers, Mapping):
return None

2
litellm/llms/anthropic/chat/guardrail_translation/handler.py
View file

@ -685,7 +685,7 @@ class AnthropicMessagesHandler(BaseTranslation):
return data
def _hoisted_top_level_system_message(self, data: dict) -> AllMessageValues | None:
def _hoisted_top_level_system_message(self, data: Mapping[str, object]) -> AllMessageValues | None:
"""Return the system message produced by translating the top-level prompt."""
system: Final = data.get("system")
if not system:

24
litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py
View file

@ -11,7 +11,6 @@ from typing import (
Final,
Literal,
Protocol,
cast, # noqa: TID251 # rebuilt message_delta dict spans the ContentBlockDelta/MessageBlockDelta union
get_args,
)
@ -27,6 +26,7 @@ from litellm.types.llms.anthropic import (
ContentBlockDelta,
ContextManagementResponse,
MessageBlockDelta,
MessageDelta,
StreamingContentBlockDeltaType,
UsageDelta,
UsageIteration,
@ -1028,26 +1028,22 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
self,
processed_chunk: ContentBlockDelta | MessageBlockDelta,
) -> ContentBlockDelta | MessageBlockDelta:
if processed_chunk.get("type") != "message_delta" or not self._refusal_text:
if processed_chunk["type"] != "message_delta" or not self._refusal_text:
return processed_chunk
delta: Final = cast(Mapping[str, object], processed_chunk["delta"]) # cast-ok: keys checked before use
delta: Final = processed_chunk["delta"]
if delta.get("stop_reason") == "max_tokens":
return processed_chunk
from litellm.llms.anthropic.experimental_pass_through.messages.utils import (
refusal_stop_details,
)
return cast( # cast-ok: rebuilt dict matches the message_delta TypedDict shape for this branch
ContentBlockDelta | MessageBlockDelta,
{ # mutable-ok: fresh translation payload; never mutated after construction
**processed_chunk,
"delta": { # mutable-ok: fresh message_delta payload; never mutated after construction
**delta,
"stop_reason": "refusal",
"stop_details": refusal_stop_details(self._refusal_text),
},
},
)
refusal_delta: Final[MessageDelta] = {
**delta,
"stop_reason": "refusal",
"stop_details": refusal_stop_details(self._refusal_text),
}
refusal_chunk: Final[MessageBlockDelta] = {**processed_chunk, "delta": refusal_delta}
return refusal_chunk
@staticmethod
def _delta_has_content(processed_chunk: Mapping[str, object]) -> bool:

2
litellm/llms/anthropic/experimental_pass_through/messages/utils.py
View file

@ -37,7 +37,7 @@ def _mapping_field(container: object, key: str) -> object | None:
"""One key of a raw provider payload, or None when the payload is not a mapping."""
if not isinstance(container, Mapping):
return None
return cast(Mapping[str, object], container).get(key) # cast-ok: raw payload, callers re-check every value
return container.get(key)
def _mapping_str_field(container: object, key: str) -> str | None:

4
litellm/llms/anthropic/experimental_pass_through/responses_adapters/transformation.py
View file

@ -169,7 +169,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
cls,
summary: Iterable[object],
encrypted_content: object,
) -> dict[str, Any] | None: # mutable-ok: API message payload
) -> dict[str, object] | None: # mutable-ok: API message payload
"""The one Anthropic block for a Responses reasoning item.
The item's encrypted reasoning rides the block's opaque field (`signature`, or
@ -198,7 +198,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
@classmethod
def _assistant_group_to_input_items(
cls, group: tuple[Mapping[str, object], ...]
) -> tuple[dict[str, Any], ...]: # mutable-ok: API message payload
) -> tuple[dict[str, object], ...]: # mutable-ok: API message payload
first: Final = group[0]
btype: Final = first.get("type")
if btype in ("thinking", "redacted_thinking"):

2
litellm/llms/openai/responses/guardrail_translation/handler.py
View file

@ -994,7 +994,7 @@ class OpenAIResponsesHandler(BaseTranslation):
def _spread_text_rewrite_over_stream_events(
self,
stream_events: Sequence[Any],
stream_events: Sequence[object],
rewritten_text: str,
guardrail_name: str,
) -> None:

7
litellm/llms/vercel_ai_gateway/embedding/transformation.py
View file

@ -7,6 +7,7 @@ Vercel AI Gateway is OpenAI-compatible and supports embeddings via the /v1/embed
Docs: https://vercel.com/docs/ai-gateway/openai-compat/embeddings
"""
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
@ -161,12 +162,14 @@ class VercelAIGatewayEmbeddingConfig(BaseEmbeddingConfig):
optional_params[param] = value
return optional_params
def get_error_class(self, error_message: str, status_code: int, headers: Any) -> BaseLLMException:
def get_error_class(
self, error_message: str, status_code: int, headers: Mapping[str, str] | httpx.Headers
) -> BaseLLMException:
"""
Get the error class for Vercel AI Gateway errors.
"""
return VercelAIGatewayException(
message=error_message,
status_code=status_code,
headers=headers,
headers=headers if isinstance(headers, httpx.Headers) else httpx.Headers(headers),
)

2
litellm/llms/vertex_ai/gemini/vertex_and_google_ai_studio_gemini.py
View file

@ -286,7 +286,7 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
"""
Check if the model is Gemini 3 or newer.
"""
model_name = model.split("/")[-1].lower()
model_name: Final = model.split("/")[-1].lower()
is_vertex_fine_tuned_model: Final = model_name.isdigit() or (
model.startswith("gemini/") and not model_name.startswith("gemini-")
)

4
litellm/main.py
View file

@ -1182,7 +1182,7 @@ def _is_claude_tool_target(custom_llm_provider: str | None, model: str) -> bool:
return False
def _without_anthropic_only_tool_keys(tool: dict) -> dict:
def _without_anthropic_only_tool_keys(tool: dict[str, object]) -> dict[str, object]:
kept: Final = {key: value for key, value in tool.items() if key not in _ANTHROPIC_ONLY_TOOL_KEYS}
function: Final = tool.get("function")
if not isinstance(function, dict):
@ -1193,7 +1193,7 @@ def _without_anthropic_only_tool_keys(tool: dict) -> dict:
}
def _drop_anthropic_only_tool_keys(tools: list[dict] | None) -> list[dict] | None:
def _drop_anthropic_only_tool_keys(tools: list[dict[str, object]] | None) -> list[dict[str, object]] | None:
if tools is None:
return None
return [_without_anthropic_only_tool_keys(tool) if isinstance(tool, dict) else tool for tool in tools]

4
litellm/passthrough/main.py
View file

@ -540,7 +540,9 @@ def llm_passthrough_route(
)
## IS STREAMING REQUEST
_streaming_request_data: dict = data if isinstance(data, dict) else (json if isinstance(json, dict) else {})
_streaming_request_data: Final[dict[str, object]] = (
data if isinstance(data, dict) else (json if isinstance(json, dict) else {})
)
is_streaming_request: Final = provider_config.is_streaming_request(
endpoint=endpoint,
request_data=_streaming_request_data,

4
litellm/proxy/_experimental/mcp_server/bridge_token_flow.py
View file

@ -86,8 +86,8 @@ async def oauth_authorization_uses_gateway_credential(request: Request) -> bool:
async def _opaque_bearer_is_gateway_credential(token: str) -> bool:
from litellm.proxy._experimental.mcp_server.outbound_credentials.envelope import (
is_envelope, # noqa: PLC0415 # envelope imports bridge types
from litellm.proxy._experimental.mcp_server.outbound_credentials.envelope import ( # noqa: PLC0415 # envelope imports bridge types
is_envelope,
is_refresh_envelope,
)
from litellm.proxy._types import hash_token # noqa: PLC0415 # proxy import cycle

4
litellm/proxy/_experimental/mcp_server/discoverable_endpoints.py
View file

@ -2634,7 +2634,9 @@ def _build_aggregate_protected_resource_response(request: Request) -> dict:
}
def _build_aggregate_authorization_server_response(request: Request, token_exchange_available: bool) -> dict:
def _build_aggregate_authorization_server_response(
request: Request, token_exchange_available: bool
) -> dict[str, object]:
"""RFC 8414 metadata for the gateway as the aggregate authorization server.
The issuer is ``{base}/mcp`` and must stay equal to the value the

2
litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
View file

@ -3490,7 +3490,7 @@ class MCPServerManager:
passthrough_server_ids: Final = [
server.server_id
for server in self.get_registry().values()
if getattr(server, "auth_type", None) == MCPAuth.true_passthrough
if server.auth_type == MCPAuth.true_passthrough
]
combined_servers.update(passthrough_server_ids)

5
litellm/proxy/_experimental/mcp_server/tool_search.py
View file

@ -128,14 +128,15 @@ def with_mcp_proxy_identity(tool: Tool, server_id: str) -> Tool:
def _mcp_proxy_identity(tool: Tool) -> MCPProxyToolIdentity:
identity: Final = (tool.meta or {}).get(_MCP_PROXY_IDENTITY_META_KEY) # mutable-ok: absent metadata default
identity: Final = None if tool.meta is None else tool.meta.get(_MCP_PROXY_IDENTITY_META_KEY)
if not isinstance(identity, Mapping):
raise TypeError("MCP proxy tool identity is missing")
server_id: Final = identity.get("server_id")
tool_name: Final = identity.get("tool_name")
if not isinstance(server_id, str) or not isinstance(tool_name, str):
raise TypeError("MCP proxy tool identity is invalid")
return {"server_id": server_id, "tool_name": tool_name} # mutable-ok: TypedDict identity payload
resolved: Final[MCPProxyToolIdentity] = {"server_id": server_id, "tool_name": tool_name}
return resolved
def mcp_proxy_tool_id(tool: Tool) -> str:

2
litellm/proxy/auth/auth_checks.py
View file

@ -4073,7 +4073,7 @@ async def get_org_object_for_request(
)
except OrganizationNotFoundError:
return None
except Exception as e: # noqa: BLE001 # only a DB outage may fail auth here, anything else degrades to no org limits
except Exception as e:
if not PrismaDBExceptionHandler.is_database_service_unavailable_error_in_chain(e):
verbose_proxy_logger.debug("org lookup failed, continuing without org limits", exc_info=True)
return None

2
litellm/proxy/common_request_processing.py
View file

@ -2323,7 +2323,7 @@ class ProxyBaseLLMRequestProcessing:
return fallbacks if isinstance(fallbacks, list) and fallbacks else None
@staticmethod
def _resolve_fallback_models(model: str, fallbacks: list) -> list | None:
def _resolve_fallback_models(model: str, fallbacks: list) -> list[str] | None:
from litellm.router_utils.fallback_event_handlers import get_fallback_model_group
fallback_model_group, generic_fallback_idx = get_fallback_model_group(

4
litellm/proxy/db/baseline_accounting.py
View file

@ -486,7 +486,7 @@ class BaselineAccountingStore:
async def _pages(
self, db: SupportsRawQueries, scope: str, after_revision: int, withdraw_from: float | None = None
) -> AsyncIterator[tuple[_StoredRecord, ...]]:
cursor: float | None = None
cursor: float | None = None # rebind-ok: keyset pagination advances after each complete timestamp group
while page := _RECORDS.validate_python(
tuple(await db.query_raw(_READ_PAGE, scope, after_revision, cursor, _PAGE_TIMESTAMPS, withdraw_from))
):
@ -627,7 +627,7 @@ async def flush_baseline_accounting(client: PrismaClient) -> None:
more_queued: Final = bool(client.baseline_accounting_transactions)
try:
remaining: Final = await asyncio.wait_for(_flush_records(store, batch), timeout=5)
except (Exception, asyncio.CancelledError) as error: # noqa: BLE001 # unknown acknowledgements can be replayed safely
except (Exception, asyncio.CancelledError) as error:
async with client.baseline_accounting_lock:
client.baseline_accounting_transactions.extend(batch)
if isinstance(error, asyncio.CancelledError):

2
litellm/proxy/guardrails/guardrail_initializers.py
View file

@ -120,7 +120,7 @@ def initialize_presidio(litellm_params: LitellmParams, guardrail: Guardrail) ->
_OPTIONAL_PresidioPIIMasking,
)
explicit_filter_scope: Final = getattr(litellm_params, "presidio_filter_scope", None)
explicit_filter_scope: Final = litellm_params.presidio_filter_scope
filter_scope: Final = explicit_filter_scope or ("input" if _is_mcp_only_mode(litellm_params.mode) else "both")
run_input: Final = filter_scope in ("input", "both")
run_output: Final = filter_scope in ("output", "both")

2
litellm/proxy/hooks/proxy_track_cost_callback.py
View file

@ -770,7 +770,7 @@ async def _reconcile_budget_reservation_before_db_update(
"Failed to invalidate budget reservation counters after pre-persist reconcile failed"
)
finally:
budget_reservation["finalized"] = True # rebind-ok: the counter update reads the stamp off the shared dict
budget_reservation["finalized"] = True # rebind-ok: stamps the caller's shared dict for the counter update
async def _release_budget_reservation(budget_reservation: dict | None) -> None:

8
litellm/proxy/management_endpoints/common_utils.py
View file

@ -528,7 +528,7 @@ def _prisma_value(value: object) -> object:
return list(value) if isinstance(value, tuple) else value
def member_budget_patch(source: BaseModel) -> dict[str, Any]:
def member_budget_patch(source: BaseModel) -> Mapping[str, object]:
"""Map the per-member limit fields a request actually set to their budget-table
columns (merge-patch: a sent value updates, an explicit null clears, an absent
field is left untouched)."""
@ -561,7 +561,7 @@ async def _upsert_budget_and_membership(
user_id: str,
existing_budget_id: str | None,
user_api_key_dict: UserAPIKeyAuth,
budget_patch: dict[str, Any],
budget_patch: Mapping[str, object],
team_default_budget_id: str | None = None,
shared_budget_ids: frozenset[str] | None = None,
):
@ -624,9 +624,9 @@ async def _upsert_budget_and_membership(
if is_shared_default and not temp_only
else None
)
source: Final[Mapping[str, Any]] = source_row.model_dump() if source_row is not None else MappingProxyType({})
source: Final[Mapping[str, object]] = source_row.model_dump() if source_row is not None else MappingProxyType({})
create_data: Final[dict[str, Any]] = { # mutable-ok: Prisma create payloads are dict-shaped
create_data: Final[dict[str, object]] = { # mutable-ok: Prisma create payloads are dict-shaped
"created_by": user_api_key_dict.user_id or "",
"updated_by": user_api_key_dict.user_id or "",
**MappingProxyType(

4
litellm/proxy/management_helpers/bulk_user_creation.py
View file

@ -348,7 +348,7 @@ async def _prepare_user(user: _PendingUser, prisma_client: PrismaClient) -> _Pre
data: Final = {**dumped, "user_id": user.user_id} # mutable-ok: /user/new defaults helper mutates in place
data_json: Final = _JSON_OBJECT.validate_python(_update_internal_new_user_params(data, user.request))
with_permission: Final = _JSON_OBJECT.validate_python(
await _set_object_permission(data_json=data_json, prisma_client=prisma_client) # pyright: ignore[reportUnknownArgumentType] # validated by the adapter
await _set_object_permission(data_json=data_json, prisma_client=prisma_client)
)
return _PreparedUser(user, _USER_ROW.validate_python(with_permission))
except Exception as exc: # noqa: BLE001 # any preparation failure is reported on this row only
@ -509,7 +509,7 @@ class _TeamsData(TypedDict):
def _default_member_budget_id(team: LiteLLM_TeamTable) -> str | None:
metadata: Final = (
_JSON_OBJECT.validate_python(
team.metadata # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # LiteLLM_TeamTable.metadata is a bare dict; validated by the adapter
team.metadata # pyright: ignore[reportUnknownMemberType] # LiteLLM_TeamTable.metadata is a bare dict; validated by the adapter
)
if team.metadata # pyright: ignore[reportUnknownMemberType] # same bare dict
else None

2
litellm/proxy/management_helpers/bulk_user_deletion.py
View file

@ -204,7 +204,7 @@ def _error_message(exc: BaseException) -> str:
if isinstance(exc, HTTPException) and isinstance(exc.detail, dict):
return str(exc.detail.get("error", exc.detail)) # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # HTTPException.detail is untyped
if isinstance(exc, HTTPException):
return str(exc.detail) # pyright: ignore[reportUnknownArgumentType] # HTTPException.detail is untyped
return str(exc.detail)
return str(exc) or type(exc).__name__

2
litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py
View file

@ -3902,7 +3902,7 @@ async def handle_gigachat_passthrough_router_model(
is_streaming: Final = request_body.get("stream", False) # pyright: ignore[reportUnknownVariableType] # request_body is dict[Unknown, Unknown]
data: dict[str, Any] = await _read_request_body(request=request) # Any needed for proxy pipeline
data: Final[dict[str, object]] = await _read_request_body(request=request)
if user_api_key_dict is not None:
auth_metadata: Final = {
metadata_key: value

4
litellm/proxy/pass_through_endpoints/llm_provider_handlers/vertex_passthrough_logging_handler.py
View file

@ -458,7 +458,7 @@ class VertexPassthroughLoggingHandler:
@staticmethod
def _is_audio_predict_response(
model: str,
json_response: dict, # mutable-ok: predicate inspects the decoded provider response dictionary without mutation
json_response: Mapping[str, object],
) -> bool:
return (
VertexPassthroughLoggingHandler._get_audio_prediction_count(json_response=json_response) > 0
@ -467,7 +467,7 @@ class VertexPassthroughLoggingHandler:
@staticmethod
def _get_audio_prediction_count(
json_response: dict, # mutable-ok: counter inspects the decoded provider response dictionary without mutation
json_response: Mapping[str, object],
) -> int:
predictions: Final = json_response.get("predictions")
if not isinstance(predictions, list):

4
litellm/proxy/response_api_endpoints/endpoints.py
View file

@ -6,7 +6,7 @@ from collections.abc import AsyncIterator, Awaitable, Mapping, Sequence
from enum import Enum
from functools import partial
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, NamedTuple, Protocol, cast, get_args
from typing import TYPE_CHECKING, Any, Final, NamedTuple, Protocol, TypeAlias, cast, get_args
from uuid import uuid4
import fastapi
@ -49,7 +49,7 @@ if TYPE_CHECKING:
router: Final = APIRouter()
_ResponseDocSchemas = dict[int | str, dict[str, Any]] # pyright: ignore[reportExplicitAny] # fastapi's responses kwarg
_ResponseDocSchemas: TypeAlias = dict[int | str, dict[str, object]] # fastapi's responses kwarg
RESPONSES_API_RESPONSE_SCHEMAS: Final[_ResponseDocSchemas] = {200: {"model": ResponsesAPIResponse}}
RESPONSES_API_CREATE_RESPONSE_SCHEMAS: Final[_ResponseDocSchemas] = {

4
litellm/proxy/spend_tracking/daily_global_spend_rollup.py
View file

@ -198,10 +198,6 @@ async def _scan_pending(prisma_client: "PrismaClient") -> _PendingScan:
return _PendingScan(marker, db_now.now, tuple(_DateRow.model_validate(row).date for row in rows))
async def pending_days(prisma_client: "PrismaClient") -> tuple[str, ...]:
return (await _scan_pending(prisma_client)).days
async def reconcile_day(prisma_client: "PrismaClient", day: str) -> None:
"""Rewrite one day of the global table from the per-key sums. Idempotent: a rerun
overwrites every group with the same totals."""

1
litellm/proxy/utils.py
View file

@ -2389,7 +2389,6 @@ class ProxyLogging:
)
try:
# Execute guardrail pipelines before the normal callback loop
if not skip_guardrails:
data, _ = await self._maybe_execute_pipelines( # rebind-ok: pipeline edits feed the callback loop below
data=data,

2
litellm/responses/litellm_completion_transformation/transformation.py
View file

@ -2254,7 +2254,7 @@ class LiteLLMCompletionResponsesConfig:
) -> Mapping[str, ResponseFunctionWebSearch]:
calls: Final[dict[str, ResponseFunctionWebSearch]] = {} # mutable-ok: indexes provider-built calls
for choice in chat_completion_response.choices:
provider_fields = getattr(choice.message, "provider_specific_fields", None)
provider_fields = choice.message.provider_specific_fields
if not isinstance(provider_fields, Mapping):
continue
web_search_calls = provider_fields.get("web_search_calls")

2
litellm/responses/streaming_iterator.py
View file

@ -1360,7 +1360,7 @@ def _billed_terminal_response(
return None
usage: Final[object] = response_obj.get("usage") # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] # a model_constructed terminal event leaves response as an untyped dict
return ResponsesAPIResponse.model_construct(
**{**response_obj, "usage": usage if usage is not None or estimate is None else estimate()} # pyright: ignore[reportUnknownArgumentType, reportArgumentType] # same untyped dict spread
**{**response_obj, "usage": usage if usage is not None or estimate is None else estimate()} # pyright: ignore[reportArgumentType] # same untyped dict spread
)

4
litellm/router_strategy/complexity_router/jev_classifier.py
View file

@ -1,7 +1,7 @@
from collections.abc import Mapping
from datetime import datetime, timezone
from types import MappingProxyType
from typing import Annotated, Final, Literal, NamedTuple, Protocol
from typing import Annotated, Final, Literal, NamedTuple, Protocol, TypeAlias
from uuid import uuid4
import httpx
@ -24,7 +24,7 @@ from litellm.proxy.pass_through_endpoints.llm_provider_handlers.typesafe_passthr
from litellm.router_strategy.complexity_router.config import DEFAULT_JEV_INSTRUCTIONS as _DEFAULT_JEV_INSTRUCTIONS
from litellm.types.utils import AUTOROUTER_CLASSIFIER_CALL_ORIGIN
JevProbability = Annotated[float, Field(ge=0.0, le=1.0)]
JevProbability: TypeAlias = Annotated[float, Field(ge=0.0, le=1.0)]
DEFAULT_JEV_INSTRUCTIONS: Final = _DEFAULT_JEV_INSTRUCTIONS

6
litellm/router_utils/pre_call_checks/encrypted_content_affinity_check.py
View file

@ -50,6 +50,7 @@ from litellm.exceptions import (
from litellm.integrations.custom_logger import CustomLogger, Span
from litellm.litellm_core_utils.credential_accessor import CredentialAccessor
from litellm.litellm_core_utils.prompt_templates.common_utils import (
anthropic_content_lists,
encrypted_content_of_block,
strip_encrypted_reasoning_from_messages,
)
@ -155,10 +156,7 @@ class EncryptedContentAffinityCheck(CustomLogger):
return iter(())
return (
cast(Mapping[str, object], block) # cast-ok: narrowed by isinstance
for message in cast(list[object], messages) # cast-ok: narrowed by isinstance
if isinstance(message, Mapping)
for content in (cast(Mapping[str, object], message).get("content"),) # cast-ok: narrowed by isinstance
if isinstance(content, list)
for content in anthropic_content_lists(cast(list[object], messages)) # cast-ok: narrowed by isinstance
for block in cast(list[object], content) # cast-ok: narrowed by isinstance
if isinstance(block, Mapping)
)

10
litellm/rust_bridge/dispatch.py
View file

@ -2,7 +2,7 @@ from __future__ import annotations
from collections.abc import Awaitable, Callable, Mapping
from dataclasses import dataclass
from typing import Final, Generic, TypeVar
from typing import Final, Generic, TypeAlias, TypeVar
from litellm.rust_bridge import catalog, runtime
from litellm.rust_bridge.bindings import NativeBinding
@ -10,11 +10,11 @@ from litellm.rust_bridge.catalog import Route, RouteContext, RouteRule, Rules
from litellm.rust_bridge.configuration import Decision
from litellm.rust_bridge.configuration import decision as rollout_decision
RequestT = TypeVar("RequestT")
NativeT = TypeVar("NativeT")
ResultT = TypeVar("ResultT")
RequestT: Final = TypeVar("RequestT")
NativeT: Final = TypeVar("NativeT")
ResultT: Final = TypeVar("ResultT")
NativeHook = Callable[[RequestT, tuple[object, ...], Mapping[str, object]], ResultT]
NativeHook: TypeAlias = Callable[[RequestT, tuple[object, ...], Mapping[str, object]], ResultT]
def call_hook(

4
litellm/rust_bridge/response_metadata.py
View file

@ -1,10 +1,10 @@
from typing import TypeVar
from typing import Final, TypeVar
from litellm.router_utils.add_retry_fallback_headers import (
_add_headers_to_response, # pyright: ignore[reportPrivateUsage] # reuse the proxy's identity-preserving response metadata writer
)
ResultT = TypeVar("ResultT")
ResultT: Final = TypeVar("ResultT")
def mark_rust_response(response: ResultT) -> ResultT:

1
litellm/types/llms/bedrock.py
View file

@ -558,7 +558,6 @@ class AmazonTitanMultimodalEmbeddingResponse(TypedDict):
message: str # Specifies any errors that occur during generation.
# TwelveLabs Marengo Embed types
TWELVELABS_EMBEDDING_INPUT_TYPES = Literal["text", "image", "video", "audio"]
TWELVELABS_EMBEDDING_OPTIONS = Literal["visual-text", "visual-image", "audio"]

4
litellm/types/utils.py
View file

@ -3860,7 +3860,7 @@ def without_server_derived_pricing(model_info: Mapping[str, Any]) -> Mapping[str
)
def echoed_cost_map_pricing_fields(model_info: Mapping[str, Any]) -> tuple[str, ...]:
def echoed_cost_map_pricing_fields(model_info: Mapping[str, object]) -> tuple[str, ...]:
"""Pricing fields a stored ``model_info`` blob copied from a ``/model/info`` response.
Only ``litellm.get_model_info`` emits ``key`` (the resolved cost-map entry), so a stored
@ -3891,7 +3891,7 @@ def echoed_cost_map_fields(
)
def pricing_override_fields(*sources: Mapping[str, Any]) -> tuple[str, ...]:
def pricing_override_fields(*sources: Mapping[str, object]) -> tuple[str, ...]:
return tuple(
sorted(
frozenset(