scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-06 08:16:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 78

fix MYPY linting

Browse source
This commit is contained in:
Ishaan Jaffer 2025-12-05 15:46:14 -08:00
parent ae065525ea
commit e519462efa
12 changed files with 68 additions and 66 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
litellm/__init__.py
View file

@ -1502,7 +1502,7 @@ def set_global_gitlab_config(config: Dict[str, Any]) -> None:
# Lazy loading system for heavy modules to reduce initial import time and memory usage
if TYPE_CHECKING:
from litellm.types.utils import ModelInfo
from litellm.types.utils import ModelInfo as _ModelInfoType
# Cost calculator functions
cost_per_token: Callable[..., Tuple[float, float]]
@ -1510,13 +1510,9 @@ if TYPE_CHECKING:
response_cost_calculator: Any
modify_integration: Any
# Utils functions - type stubs for lazy loaded functions
exception_type: Callable[..., Any]
get_optional_params: Callable[..., dict]
# Utils functions - type stubs for truly lazy loaded functions only
# (functions NOT imported via "from .main import *")
get_response_string: Callable[..., str]
token_counter: Callable[..., int]
create_pretrained_tokenizer: Callable[..., Any]
create_tokenizer: Callable[..., Any]
supports_function_calling: Callable[..., bool]
supports_web_search: Callable[..., bool]
supports_url_context: Callable[..., bool]
@ -1527,10 +1523,9 @@ if TYPE_CHECKING:
supports_audio_output: Callable[..., bool]
supports_system_messages: Callable[..., bool]
supports_reasoning: Callable[..., bool]
get_litellm_params: Callable[..., dict]
acreate: Callable[..., Any]
get_max_tokens: Callable[..., int]
get_model_info: Callable[..., ModelInfo]
get_model_info: Callable[..., _ModelInfoType]
register_prompt_template: Callable[..., None]
validate_environment: Callable[..., dict]
check_valid_key: Callable[..., bool]
@ -1538,22 +1533,15 @@ if TYPE_CHECKING:
encode: Callable[..., list]
decode: Callable[..., str]
_calculate_retry_after: Callable[..., float]
_should_retry: Callable[[int], bool]
_should_retry: Callable[..., bool]
get_supported_openai_params: Callable[..., Optional[list]]
get_api_base: Callable[..., Optional[str]]
get_first_chars_messages: Callable[..., str]
get_provider_fields: Callable[..., dict]
get_provider_fields: Callable[..., List]
get_valid_models: Callable[..., list]
# Response types - lazy loaded
ModelResponse: Type[Any]
ModelResponseStream: Type[Any]
EmbeddingResponse: Type[Any]
ImageResponse: Type[Any]
TranscriptionResponse: Type[Any]
TextCompletionResponse: Type[Any]
# Response types - truly lazy loaded only (not in main.py or elsewhere)
ModelResponseListIterator: Type[Any]
Logging: Type[Any]
def __getattr__(name: str) -> Any:

4
litellm/litellm_core_utils/get_llm_provider_logic.py
View file

@ -469,11 +469,13 @@ def _get_openai_compatible_provider_info( # noqa: PLR0915
model = model.split("/", 1)[1]
# Check JSON providers FIRST (before hardcoded ones)
from litellm.llms.openai_like.json_loader import JSONProviderRegistry
from litellm.llms.openai_like.dynamic_config import create_config_class
from litellm.llms.openai_like.json_loader import JSONProviderRegistry
if JSONProviderRegistry.exists(custom_llm_provider):
provider_config = JSONProviderRegistry.get(custom_llm_provider)
if provider_config is None:
raise ValueError(f"Provider {custom_llm_provider} not found")
config_class = create_config_class(provider_config)
api_base, dynamic_api_key = config_class()._get_openai_compatible_provider_info(
api_base, api_key

46
litellm/llms/openai/chat/guardrail_translation/handler.py
View file

@ -164,7 +164,7 @@ class OpenAIChatCompletionsHandler(BaseTranslation):
for tool_call_idx, tool_call in enumerate(tool_calls):
if isinstance(tool_call, dict):
# Add the full tool call object to the list
tool_calls_to_check.append(ChatCompletionToolParam(**tool_call))
tool_calls_to_check.append(cast(ChatCompletionToolParam, tool_call))
tool_call_task_mappings.append((msg_idx, int(tool_call_idx)))
async def _apply_guardrail_responses_to_input_texts(
@ -380,20 +380,20 @@ class OpenAIChatCompletionsHandler(BaseTranslation):
if isinstance(content, str):
# String content - accumulate for this choice
key = (choice_idx, None)
if key not in combined_texts:
combined_texts[key] = ""
combined_texts[key] += content
str_key: Tuple[int, Optional[int]] = (choice_idx, None)
if str_key not in combined_texts:
combined_texts[str_key] = ""
combined_texts[str_key] += content
elif isinstance(content, list):
# List content - accumulate for each content item
for content_idx, content_item in enumerate(content):
text_str = content_item.get("text")
if text_str:
key = (choice_idx, content_idx)
if key not in combined_texts:
combined_texts[key] = ""
combined_texts[key] += text_str
list_key: Tuple[int, Optional[int]] = (choice_idx, content_idx)
if list_key not in combined_texts:
combined_texts[list_key] = ""
combined_texts[list_key] += text_str
# Step 2: Create lists for guardrail processing
texts_to_check: List[str] = []
@ -401,9 +401,9 @@ class OpenAIChatCompletionsHandler(BaseTranslation):
task_mappings: List[Tuple[int, Optional[int]]] = []
# Track (choice_index, content_index) for each combined text
for (choice_idx, content_idx), combined_text in combined_texts.items():
for (map_choice_idx, map_content_idx), combined_text in combined_texts.items():
texts_to_check.append(combined_text)
task_mappings.append((choice_idx, content_idx))
task_mappings.append((map_choice_idx, map_content_idx))
# Step 3: Apply guardrail to all combined texts in batch
if texts_to_check:
@ -503,7 +503,7 @@ class OpenAIChatCompletionsHandler(BaseTranslation):
# Determine content source and tool calls based on choice type
content = None
tool_calls = None
tool_calls: Optional[List[Any]] = None
if isinstance(choice, litellm.Choices):
content = choice.message.content
tool_calls = choice.message.tool_calls
@ -686,15 +686,15 @@ class OpenAIChatCompletionsHandler(BaseTranslation):
if isinstance(content, str):
# String content
key = (choice_idx_in_response, None)
if key in guardrail_map:
if key not in already_set:
str_key: Tuple[int, Optional[int]] = (choice_idx_in_response, None)
if str_key in guardrail_map:
if str_key not in already_set:
# First chunk - set the complete guardrailed text
if isinstance(choice, litellm.StreamingChoices):
choice.delta.content = guardrail_map[key]
choice.delta.content = guardrail_map[str_key]
elif isinstance(choice, litellm.Choices):
choice.message.content = guardrail_map[key]
already_set[key] = True
choice.message.content = guardrail_map[str_key]
already_set[str_key] = True
else:
# Subsequent chunks - clear the content
if isinstance(choice, litellm.StreamingChoices):
@ -706,12 +706,12 @@ class OpenAIChatCompletionsHandler(BaseTranslation):
# List content - handle each content item
for content_idx, content_item in enumerate(content):
if "text" in content_item:
key = (choice_idx_in_response, content_idx)
if key in guardrail_map:
if key not in already_set:
list_key: Tuple[int, Optional[int]] = (choice_idx_in_response, content_idx)
if list_key in guardrail_map:
if list_key not in already_set:
# First chunk - set the complete guardrailed text
content_item["text"] = guardrail_map[key]
already_set[key] = True
content_item["text"] = guardrail_map[list_key]
already_set[list_key] = True
else:
# Subsequent chunks - clear the text
content_item["text"] = ""

7
litellm/llms/openai_like/dynamic_config.py
View file

@ -19,11 +19,11 @@ def create_config_class(provider: SimpleProviderConfig):
"""Generate config class dynamically from JSON configuration"""
# Choose base class
base_class = (
base_class: type = (
OpenAIGPTConfig if provider.base_class == "openai_gpt" else OpenAILikeChatConfig
)
class JSONProviderConfig(base_class):
class JSONProviderConfig(base_class): # type: ignore[valid-type,misc]
@overload
def _transform_messages(
self, messages: List[AllMessageValues], model: str, is_async: Literal[True]
@ -87,6 +87,9 @@ def create_config_class(provider: SimpleProviderConfig):
if not api_base:
api_base = provider.base_url
if api_base is None:
raise ValueError(f"api_base is required for provider {provider.slug}")
if not api_base.endswith("/chat/completions"):
api_base = f"{api_base}/chat/completions"

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

@ -116,7 +116,7 @@ def _process_gemini_image(
is not None
):
file_data = FileDataType(file_uri=image_url, mime_type=image_type)
part: PartType = {"file_data": file_data}
part = {"file_data": file_data}
if media_resolution_enum is not None and model is not None:
from .vertex_and_google_ai_studio_gemini import VertexGeminiConfig
@ -129,7 +129,7 @@ def _process_gemini_image(
image = convert_to_anthropic_image_obj(image_url, format=format)
_blob: BlobType = {"data": image["data"], "mime_type": image["media_type"]}
part: PartType = {"inline_data": cast(BlobType, _blob)}
part = {"inline_data": cast(BlobType, _blob)}
if media_resolution_enum is not None and model is not None:
from .vertex_and_google_ai_studio_gemini import VertexGeminiConfig

2
litellm/llms/vertex_ai/text_to_speech/transformation.py
View file

@ -220,7 +220,7 @@ class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
Returns:
Tuple of (mapped_voice_str, mapped_params)
"""
mapped_params = {}
mapped_params: Dict[str, Any] = {}
##########################################################
# Map voice using helper

14
litellm/proxy/auth/login_utils.py
View file

@ -9,9 +9,9 @@ import os
import secrets
from typing import Literal, Optional, cast
import litellm
from fastapi import HTTPException
import litellm
from litellm.constants import LITELLM_PROXY_ADMIN_NAME
from litellm.proxy._types import (
LiteLLM_UserTable,
@ -64,13 +64,19 @@ def get_ui_credentials(master_key: Optional[str]) -> tuple[str, str]:
class LoginResult:
"""Result object containing authentication data from login."""
user_id: str
key: str
user_email: Optional[str]
user_role: str
login_method: Literal["sso", "username_password"]
def __init__(
self,
user_id: str,
key: str,
user_email: Optional[str],
user_role: str,
login_method: str = "username_password",
login_method: Literal["sso", "username_password"] = "username_password",
):
self.user_id = user_id
self.key = key
@ -193,14 +199,14 @@ async def authenticate_user(
key = response["token"] # type: ignore
if get_secret_bool("EXPERIMENTAL_UI_LOGIN"):
from litellm.proxy.auth.auth_checks import ExperimentalUIJWTToken
user_info: Optional[LiteLLM_UserTable] = None
if _user_row is not None:
user_info = _user_row
elif (
user_id is not None
): # if user_id is not None, we are using the UI_USERNAME and UI_PASSWORD
from litellm.proxy.auth.auth_checks import ExperimentalUIJWTToken
user_info = LiteLLM_UserTable(
user_id=user_id,
user_role=user_role,

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

@ -127,7 +127,7 @@ class GenericGuardrailAPI(CustomGuardrail):
for field_name in GenericGuardrailAPIMetadata.__annotations__.keys():
value = metadata_dict.get(field_name)
if value is not None:
result_metadata[field_name] = value
result_metadata[field_name] = value # type: ignore[literal-required]
# handle user_api_key_token = user_api_key_hash
if metadata_dict.get("user_api_key_token") is not None:

2
litellm/proxy/health_endpoints/_health_endpoints.py
View file

@ -409,7 +409,7 @@ def _build_model_param_to_info_mapping(model_list: list) -> dict:
Returns:
Dictionary mapping model parameter to list of model info dicts
"""
model_param_to_info = {}
model_param_to_info: dict = {}
for model in model_list:
model_info = model.get("model_info", {})
model_name = model.get("model_name")

1
litellm/proxy/hooks/key_management_event_hooks.py
View file

@ -7,7 +7,6 @@ import litellm
from litellm._logging import verbose_proxy_logger
from litellm._uuid import uuid
from litellm.proxy._types import (
CommonProxyErrors,
GenerateKeyRequest,
GenerateKeyResponse,
KeyRequest,

22
litellm/proxy/proxy_server.py
View file

@ -45,7 +45,10 @@ from litellm.constants import (
LITELLM_SETTINGS_SAFE_DB_OVERRIDES,
)
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
from litellm.proxy.common_utils.callback_utils import normalize_callback_names
from litellm.proxy.common_utils.callback_utils import (
normalize_callback_names,
process_callback,
)
from litellm.proxy.common_utils.realtime_utils import _realtime_request_body
from litellm.types.utils import (
ModelResponse,
@ -54,7 +57,6 @@ from litellm.types.utils import (
TokenCountResponse,
)
from litellm.utils import load_credentials_from_list
from litellm.proxy.common_utils.callback_utils import process_callback
if TYPE_CHECKING:
from aiohttp import ClientSession
@ -168,8 +170,8 @@ from litellm.constants import (
PROXY_BUDGET_RESCHEDULER_MIN_TIME,
)
from litellm.exceptions import RejectedRequestError
from litellm.integrations.SlackAlerting.slack_alerting import SlackAlerting
from litellm.integrations.custom_logger import CustomLogger
from litellm.integrations.SlackAlerting.slack_alerting import SlackAlerting
from litellm.litellm_core_utils.core_helpers import (
_get_parent_otel_span_from_kwargs,
get_litellm_metadata_from_kwargs,
@ -613,7 +615,7 @@ async def proxy_shutdown_event():
await jwt_handler.close()
if db_writer_client is not None:
await db_writer_client.close()
await db_writer_client.close() # type: ignore[reportGeneralTypeIssues]
# flush remaining langfuse logs
if "langfuse" in litellm.success_callback:
@ -792,7 +794,7 @@ async def proxy_startup_event(app: FastAPI):
except Exception as e:
verbose_proxy_logger.error(f"Error closing shared aiohttp session: {e}")
await proxy_shutdown_event()
await proxy_shutdown_event() # type: ignore[reportGeneralTypeIssues]
app = FastAPI(
@ -802,7 +804,7 @@ app = FastAPI(
description=_description,
version=version,
root_path=server_root_path, # check if user passed root path, FastAPI defaults this value to ""
lifespan=proxy_startup_event,
lifespan=proxy_startup_event, # type: ignore[reportGeneralTypeIssues]
)
vertex_live_passthrough_vertex_base = VertexBase()
@ -8330,9 +8332,9 @@ async def login(request: Request): # noqa: PLR0915
# Generate JWT token
import jwt
jwt_token = jwt.encode( # type: ignore
jwt_token = jwt.encode(
cast(dict, returned_ui_token_object),
master_key,
cast(str, master_key),
algorithm="HS256",
)
@ -8377,9 +8379,9 @@ async def login_v2(request: Request): # noqa: PLR0915
import jwt
jwt_token = jwt.encode( # type: ignore
jwt_token = jwt.encode(
cast(dict, returned_ui_token_object),
master_key,
cast(str, master_key),
algorithm="HS256",
)

4
litellm/utils.py
View file

@ -7023,11 +7023,13 @@ class ProviderConfigManager:
"""
# Check JSON providers FIRST
from litellm.llms.openai_like.json_loader import JSONProviderRegistry
from litellm.llms.openai_like.dynamic_config import create_config_class
from litellm.llms.openai_like.json_loader import JSONProviderRegistry
if JSONProviderRegistry.exists(provider.value):
provider_config = JSONProviderRegistry.get(provider.value)
if provider_config is None:
raise ValueError(f"Provider {provider.value} not found")
return create_config_class(provider_config)()
if (