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>

litellm/integrations/mavvrik_focus/mavvrik_focus_logger.py

@@ -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")

litellm/litellm_core_utils/core_helpers.py

@@ -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

litellm/litellm_core_utils/llm_cost_calc/zero_cost_diagnostic.py

@@ -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
 
 

litellm/llms/vercel_ai_gateway/embedding/transformation.py

@@ -161,7 +161,7 @@ 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: dict | httpx.Headers) -> BaseLLMException:
         """
         Get the error class for Vercel AI Gateway errors.
         """