refactor(realtime): align annotations with current lint rules

litellm/litellm_core_utils/realtime_streaming.py

@@ -524,9 +524,8 @@ class RealTimeStreaming:
         for event in self.messages:
             if event.get("type") != "session.closed":
                 continue
-            event_usage = event.get("usage")  # rebind-ok: each close event carries independent usage
             if (
-                isinstance(event_usage, dict)
+                isinstance(event_usage := event.get("usage"), dict)
                 and normalized_audio_duration_seconds(event_usage.get("output_seconds")) is not None
             ):
                 self._translation_usage_finalized = True

litellm/llms/custom_httpx/llm_http_handler.py

@@ -6162,7 +6162,7 @@ class BaseLLMHTTPHandler:
                 configured_client: Final = openai_client.with_options(
                     timeout=timeout,
                     set_default_headers={  # mutable-ok: OpenAI SDK accepts a mutable custom-header mapping
-                        key: str(value)  # mutable-ok: SDK headers are materialized as a concrete string mapping
+                        key: str(value)
                         for key, value in (extra_headers or {}).items()  # mutable-ok: SDK requires concrete headers
                     },
                 )

litellm/llms/openai/transcriptions/gpt_transformation.py

@@ -40,9 +40,7 @@ class OpenAIGPTAudioTranscriptionConfig(OpenAIWhisperAudioTranscriptionConfig):
 
 
 class OpenAIGPTTranscribeAudioTranscriptionConfig(OpenAIGPTAudioTranscriptionConfig):
-    def get_supported_openai_params(  # mutable-ok: base transcription interface returns a mutable parameter list
-        self, model: str
-    ) -> list[OpenAIAudioTranscriptionOptionalParams]:
+    def get_supported_openai_params(self, model: str) -> list[OpenAIAudioTranscriptionOptionalParams]:
         return [
             "prompt",
             "response_format",

litellm/llms/openai/transcriptions/handler.py

@@ -57,7 +57,9 @@ class OpenAIAudioTranscription(OpenAIChatCompletion):
             if data.get("stream") is True:
                 stream_response: Final = openai_client.audio.transcriptions.create(**sdk_data, timeout=timeout)
                 return None, stream_response
-            raw_response: Final = openai_client.audio.transcriptions.with_raw_response.create(**sdk_data, timeout=timeout)
+            raw_response: Final = openai_client.audio.transcriptions.with_raw_response.create(
+                **sdk_data, timeout=timeout
+            )
             headers: Final = dict(raw_response.headers)
             response: Final = raw_response.parse()
             return headers, response

litellm/proxy/realtime_endpoints/endpoints.py

@@ -139,7 +139,7 @@ async def _authorize_and_bind_nested_transcription_models(
     session_data: dict,  # mutable-ok: session payload is rewritten in place for provider serialization
     user_api_key_dict: UserAPIKeyAuth,
     llm_model_list: list | None,  # mutable-ok: inherited auth helper accepts the proxy model list
-    llm_router: Any,
+    llm_router: "Router | None",
 ) -> None:
     nested_models: Final = tuple(_transcription_model_candidates_from_session(session_data))
     for nested_model in nested_models:
@@ -515,18 +515,14 @@ async def proxy_realtime_calls(
         session_type = _coerce_realtime_session_type(raw_session_type)
         if is_translation_request != (raw_session_type == "translation"):
             return Response(
-                content=json.dumps(  # mutable-ok: JSON encoder requires the endpoint error payload mapping
-                    {"error": "Token is not valid for this Realtime endpoint"}
-                ),
+                content=json.dumps({"error": "Token is not valid for this Realtime endpoint"}),
                 status_code=http_status.HTTP_401_UNAUTHORIZED,
                 media_type="application/json",
             )
     else:
         if is_translation_request:
             return Response(
-                content=json.dumps(  # mutable-ok: JSON encoder requires the endpoint error payload mapping
-                    {"error": "Token is not valid for this Realtime endpoint"}
-                ),
+                content=json.dumps({"error": "Token is not valid for this Realtime endpoint"}),
                 status_code=http_status.HTTP_401_UNAUTHORIZED,
                 media_type="application/json",
             )

litellm/realtime_api/main.py

@@ -75,12 +75,14 @@ def _with_resolved_session_model(session: dict[str, object], model_name: str) ->
         audio = session.get("audio")
         audio = audio if isinstance(audio, dict) else {}  # mutable-ok: nested session model is rebuilt locally
         audio_input = audio.get("input")
-        audio_input = (  # mutable-ok: nested session model is rebuilt locally
-            audio_input if isinstance(audio_input, dict) else {}
+        audio_input = (
+            audio_input if isinstance(audio_input, dict) else {}  # mutable-ok: nested session model is rebuilt locally
         )
         transcription = audio_input.get("transcription")
-        transcription = (  # mutable-ok: nested session model is rebuilt locally
-            transcription if isinstance(transcription, dict) else {}
+        transcription = (
+            transcription
+            if isinstance(transcription, dict)
+            else {}  # mutable-ok: nested session model is rebuilt locally
         )
         return {  # mutable-ok: provider routing requires an independently mutable session payload
             **session,