fix(mypy): resolve type errors across 9 files

- batches/main.py: import FileExpiresAfter, cast output_expires_after on assignment
- openai/openai.py, azure/batches/handler.py: add # type: ignore[arg-type] on
  batches.create / batches.retrieve TypedDict unpacking calls
- searchapi/transformation.py: cast optional_params["country"] to str before .lower()
- openrouter/image_edit/transformation.py: cast iterated value to str for size/quality params
- spend_log_cleanup.py: narrow bool | None to bool with `or False`
- cost_tracking_settings.py: cast base_model/resolved_model to str and
  custom_llm_provider to Optional[str] in return statements
- text_moderation.py: suppress misc TypedDict ** expansion error; use cast for response
- prompt_shield.py: use cast instead of TypedDict(**response_json) construction

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

litellm/batches/main.py

@@ -33,6 +33,7 @@ from litellm.secret_managers.main import get_secret_str
 from litellm.types.llms.openai import (
     CancelBatchRequest,
     CreateBatchRequest,
+    FileExpiresAfter,
     RetrieveBatchRequest,
 )
 from litellm.types.router import GenericLiteLLMParams
@@ -219,7 +220,7 @@ def create_batch(  # noqa: PLR0915
             extra_body=extra_body,
         )
         if output_expires_after is not None:
-            _create_batch_request["output_expires_after"] = output_expires_after
+            _create_batch_request["output_expires_after"] = cast(FileExpiresAfter, output_expires_after)
         if model is not None:
             provider_config = ProviderConfigManager.get_provider_batches_config(
                 model=model,

litellm/llms/azure/batches/handler.py

@@ -35,7 +35,7 @@ class AzureBatchesAPI(BaseAzureLLM):
         create_batch_data: CreateBatchRequest,
         azure_client: Union[AsyncAzureOpenAI, AsyncOpenAI],
     ) -> LiteLLMBatch:
-        response = await azure_client.batches.create(**create_batch_data)
+        response = await azure_client.batches.create(**create_batch_data)  # type: ignore[arg-type]
         return LiteLLMBatch(**response.model_dump())
 
     def create_batch(
@@ -73,7 +73,7 @@ class AzureBatchesAPI(BaseAzureLLM):
             return self.acreate_batch(  # type: ignore
                 create_batch_data=create_batch_data, azure_client=azure_client
             )
-        response = cast(Union[AzureOpenAI, OpenAI], azure_client).batches.create(**create_batch_data)
+        response = cast(Union[AzureOpenAI, OpenAI], azure_client).batches.create(**create_batch_data)  # type: ignore[arg-type]
         return LiteLLMBatch(**response.model_dump())
 
     async def aretrieve_batch(
@@ -81,7 +81,7 @@ class AzureBatchesAPI(BaseAzureLLM):
         retrieve_batch_data: RetrieveBatchRequest,
         client: Union[AsyncAzureOpenAI, AsyncOpenAI],
     ) -> LiteLLMBatch:
-        response = await client.batches.retrieve(**retrieve_batch_data)
+        response = await client.batches.retrieve(**retrieve_batch_data)  # type: ignore[arg-type]
         return LiteLLMBatch(**response.model_dump())
 
     def retrieve_batch(

litellm/llms/openai/openai.py

@@ -1938,7 +1938,7 @@ class OpenAIBatchesAPI(BaseLLM):
         create_batch_data: CreateBatchRequest,
         openai_client: AsyncOpenAI,
     ) -> LiteLLMBatch:
-        response = await openai_client.batches.create(**create_batch_data)
+        response = await openai_client.batches.create(**create_batch_data)  # type: ignore[arg-type]
         return LiteLLMBatch(**response.model_dump())
 
     def create_batch(
@@ -1974,7 +1974,7 @@ class OpenAIBatchesAPI(BaseLLM):
             return self.acreate_batch(  # type: ignore
                 create_batch_data=create_batch_data, openai_client=openai_client
             )
-        response = cast(OpenAI, openai_client).batches.create(**create_batch_data)
+        response = cast(OpenAI, openai_client).batches.create(**create_batch_data)  # type: ignore[arg-type]
 
         return LiteLLMBatch(**response.model_dump())
 
@@ -1984,7 +1984,7 @@ class OpenAIBatchesAPI(BaseLLM):
         openai_client: AsyncOpenAI,
     ) -> LiteLLMBatch:
         verbose_logger.debug("retrieving batch, args= %s", retrieve_batch_data)
-        response = await openai_client.batches.retrieve(**retrieve_batch_data)
+        response = await openai_client.batches.retrieve(**retrieve_batch_data)  # type: ignore[arg-type]
         return LiteLLMBatch(**response.model_dump())
 
     def retrieve_batch(
@@ -2020,7 +2020,7 @@ class OpenAIBatchesAPI(BaseLLM):
             return self.aretrieve_batch(  # type: ignore
                 retrieve_batch_data=retrieve_batch_data, openai_client=openai_client
             )
-        response = cast(OpenAI, openai_client).batches.retrieve(**retrieve_batch_data)
+        response = cast(OpenAI, openai_client).batches.retrieve(**retrieve_batch_data)  # type: ignore[arg-type]
         return LiteLLMBatch(**response.model_dump())
 
     async def acancel_batch(

litellm/llms/openrouter/image_edit/transformation.py

@@ -91,9 +91,9 @@ class OpenRouterImageEditConfig(BaseImageEditConfig):
                 if key == "size":
                     if "image_config" not in mapped_params:
                         mapped_params["image_config"] = {}
-                    mapped_params["image_config"]["aspect_ratio"] = self._map_size_to_aspect_ratio(value)
+                    mapped_params["image_config"]["aspect_ratio"] = self._map_size_to_aspect_ratio(cast(str, value))
                 elif key == "quality":
-                    image_size = self._map_quality_to_image_size(value)
+                    image_size = self._map_quality_to_image_size(cast(str, value))
                     if image_size:
                         if "image_config" not in mapped_params:
                             mapped_params["image_config"] = {}

litellm/llms/searchapi/search/transformation.py

@@ -3,7 +3,7 @@ Calls SearchAPI.io's Google Search API endpoint.
 
 SearchAPI.io API Reference: https://www.searchapi.io/docs/google
 """
-from typing import Dict, List, Literal, Optional, TypedDict, Union
+from typing import Dict, List, Literal, Optional, TypedDict, Union, cast
 from urllib.parse import urlencode
 
 import httpx
@@ -164,7 +164,7 @@ class SearchAPIConfig(BaseSearchConfig):
 
         if "country" in optional_params:
             # Map to gl parameter
-            result_data["gl"] = optional_params["country"].lower()
+            result_data["gl"] = cast(str, optional_params["country"]).lower()
 
         # Pass through all other SearchAPI.io-specific parameters
         for param, value in optional_params.items():

litellm/proxy/db/db_transaction_queue/spend_log_cleanup.py

@@ -133,7 +133,7 @@ class SpendLogCleanup:
             if self.pod_lock_manager and self.pod_lock_manager.redis_cache:
                 lock_acquired = await self.pod_lock_manager.acquire_lock(
                     cronjob_id=SPEND_LOG_CLEANUP_JOB_NAME,
-                )
+                ) or False
                 verbose_proxy_logger.info(
                     f"Lock acquisition attempt: {'successful' if lock_acquired else 'failed'}  at {datetime.now()}"
                 )

litellm/proxy/guardrails/guardrail_hooks/azure/prompt_shield.py

@@ -98,7 +98,7 @@ class AzureContentSafetyPromptShieldGuardrail(AzureGuardrailBase, CustomGuardrai
                 "text:shieldPrompt", cast(dict, request_body)
             )
 
-            last_response = AzurePromptShieldGuardrailResponse(**response_json)
+            last_response = cast(AzurePromptShieldGuardrailResponse, response_json)
 
             if last_response["userPromptAnalysis"].get("attackDetected"):
                 verbose_proxy_logger.warning(

litellm/proxy/guardrails/guardrail_hooks/azure/text_moderation.py

@@ -125,13 +125,13 @@ class AzureContentSafetyTextModerationGuardrail(AzureGuardrailBase, CustomGuardr
         for chunk in chunks:
             request_body = AzureTextModerationGuardrailRequestBody(
                 text=chunk,
-                **self.optional_params_request_body,
+                **self.optional_params_request_body,  # type: ignore[misc]
             )
             response_json = await self._post_to_content_safety(
                 "text:analyze", cast(dict, request_body)
             )
 
-            chunk_response = AzureTextModerationGuardrailResponse(**response_json)
+            chunk_response = cast(AzureTextModerationGuardrailResponse, response_json)
 
             # For multi-chunk texts the callers only see the final response,
             # so we must check every intermediate chunk here to avoid silently

litellm/proxy/management_endpoints/cost_tracking_settings.py

@@ -10,7 +10,7 @@ PATCH /config/cost_margin_config - Update cost margin configuration
 POST /cost/estimate - Estimate cost for a given model and token counts
 """
 
-from typing import Dict, Optional, Tuple, Union
+from typing import Dict, Optional, Tuple, Union, cast
 
 from fastapi import APIRouter, Depends, HTTPException
 
@@ -66,8 +66,8 @@ def _resolve_model_for_cost_lookup(model: str) -> Tuple[str, Optional[str]]:
                     verbose_proxy_logger.debug(
                         f"Resolved model '{model}' to base_model '{base_model}' from router"
                     )
-                    custom_llm_provider = litellm_params.get("custom_llm_provider")
-                    return base_model, custom_llm_provider
+                    custom_llm_provider = cast(Optional[str], litellm_params.get("custom_llm_provider"))
+                    return cast(str, base_model), custom_llm_provider
 
                 resolved_model = litellm_params.get("model")
 
@@ -75,8 +75,8 @@ def _resolve_model_for_cost_lookup(model: str) -> Tuple[str, Optional[str]]:
                     verbose_proxy_logger.debug(
                         f"Resolved model '{model}' to '{resolved_model}' from router"
                     )
-                    custom_llm_provider = litellm_params.get("custom_llm_provider")
-                    return resolved_model, custom_llm_provider
+                    custom_llm_provider = cast(Optional[str], litellm_params.get("custom_llm_provider"))
+                    return cast(str, resolved_model), custom_llm_provider
         except Exception as e:
             verbose_proxy_logger.debug(
                 f"Could not resolve model '{model}' from router: {e}"