diff --git a/litellm/llms/gigachat/__init__.py b/litellm/llms/gigachat/__init__.py index af5d2717643..e7c2206ffaa 100644 --- a/litellm/llms/gigachat/__init__.py +++ b/litellm/llms/gigachat/__init__.py @@ -17,9 +17,9 @@ from .chat.transformation import GigaChatConfig, GigaChatError from .embedding.transformation import GigaChatEmbeddingConfig from .passthrough.transformation import GigaChatPassthroughConfig -__all__ = [ +__all__ = ( "GigaChatConfig", "GigaChatEmbeddingConfig", "GigaChatError", "GigaChatPassthroughConfig", -] +) diff --git a/litellm/llms/gigachat/authenticator.py b/litellm/llms/gigachat/authenticator.py index 01272c42284..a1c6bda093f 100644 --- a/litellm/llms/gigachat/authenticator.py +++ b/litellm/llms/gigachat/authenticator.py @@ -7,6 +7,7 @@ Based on official GigaChat SDK authentication flow. import time import uuid +from collections.abc import Mapping from typing import Final import httpx @@ -63,7 +64,7 @@ def get_access_token( credentials: str | None = None, scope: str | None = None, auth_url: str | None = None, - litellm_params: dict | None = None, + litellm_params: Mapping[str, object] | None = None, ) -> str: """ Get valid access token, using cache if available. @@ -80,26 +81,28 @@ def get_access_token( GigaChatAuthError: If authentication fails """ if not litellm_params: - litellm_params = {} + litellm_params = {} # mutable-ok: empty dict default; rebind-ok: provide default - access_token = litellm_params.get("gigachat_access_token") or get_secret_str("GIGACHAT_ACCESS_TOKEN") + access_token: Final = litellm_params.get("gigachat_access_token") or get_secret_str("GIGACHAT_ACCESS_TOKEN") if access_token: return access_token - credentials = credentials or _get_credentials() - if not credentials: + effective_credentials: Final = credentials or _get_credentials() + if not effective_credentials: raise GigaChatAuthError( status_code=401, message="GigaChat credentials not provided. Set GIGACHAT_CREDENTIALS or GIGACHAT_API_KEY environment variable.", ) - scope = scope or litellm_params.get("gigachat_scope") or _get_scope() - auth_url = auth_url or litellm_params.get("gigachat_auth_url") or _get_auth_url() + effective_scope: Final = scope or litellm_params.get("gigachat_scope") or _get_scope() + effective_auth_url: Final = auth_url or litellm_params.get("gigachat_auth_url") or _get_auth_url() # Check cache - cache_key: Final = f"gigachat_token:{credentials[:16]}" + cache_key: Final = f"gigachat_token:{effective_credentials[:16]}" cached: Final = _token_cache.get_cache(cache_key) if cached: + token: Final + expires_at: Final token, expires_at = cached # Check if token is still valid (with buffer) if time.time() * 1000 < expires_at - TOKEN_EXPIRY_BUFFER_MS: @@ -107,7 +110,7 @@ def get_access_token( return token # Request new token - token, expires_at = _request_token_sync(credentials, scope, auth_url) + token, expires_at = _request_token_sync(effective_credentials, effective_scope, effective_auth_url) if expires_at: # Cache token @@ -122,37 +125,39 @@ async def get_access_token_async( credentials: str | None = None, scope: str | None = None, auth_url: str | None = None, - litellm_params: dict | None = None, + litellm_params: Mapping[str, object] | None = None, ) -> str: """Async version of get_access_token.""" if not litellm_params: - litellm_params = {} + litellm_params = {} # mutable-ok: empty dict default; rebind-ok: provide default - access_token = litellm_params.get("gigachat_access_token") or get_secret_str("GIGACHAT_ACCESS_TOKEN") + access_token: Final = litellm_params.get("gigachat_access_token") or get_secret_str("GIGACHAT_ACCESS_TOKEN") if access_token: return access_token - credentials = credentials or _get_credentials() - if not credentials: + effective_credentials: Final = credentials or _get_credentials() + if not effective_credentials: raise GigaChatAuthError( status_code=401, message="GigaChat credentials not provided. Set GIGACHAT_CREDENTIALS or GIGACHAT_API_KEY environment variable.", ) - scope = scope or litellm_params.get("gigachat_scope") or _get_scope() - auth_url = auth_url or litellm_params.get("gigachat_auth_url") or _get_auth_url() + effective_scope: Final = scope or litellm_params.get("gigachat_scope") or _get_scope() + effective_auth_url: Final = auth_url or litellm_params.get("gigachat_auth_url") or _get_auth_url() # Check cache - cache_key: Final = f"gigachat_token:{credentials[:16]}" + cache_key: Final = f"gigachat_token:{effective_credentials[:16]}" cached: Final = _token_cache.get_cache(cache_key) if cached: + token: Final + expires_at: Final token, expires_at = cached if time.time() * 1000 < expires_at - TOKEN_EXPIRY_BUFFER_MS: verbose_logger.debug("Using cached GigaChat access token") return token # Request new token - token, expires_at = await _request_token_async(credentials, scope, auth_url) + token, expires_at = await _request_token_async(effective_credentials, effective_scope, effective_auth_url) if expires_at: # Cache token @@ -241,7 +246,7 @@ def _parse_token_response(response: httpx.Response) -> tuple[str, int]: # GigaChat returns either 'tok'/'exp' or 'access_token'/'expires_at' access_token: Final = data.get("tok") or data.get("access_token") - expires_at = data.get("exp") or data.get("expires_at") + expires_at_raw: Final = data.get("exp") or data.get("expires_at") if not access_token: raise GigaChatAuthError( @@ -249,9 +254,12 @@ def _parse_token_response(response: httpx.Response) -> tuple[str, int]: message=f"Invalid token response: {data}", ) + expires_at: int # expires_at is in milliseconds - if isinstance(expires_at, str): - expires_at = int(expires_at) + if isinstance(expires_at_raw, str): + expires_at = int(expires_at_raw) + else: + expires_at = expires_at_raw # pyright: ignore[reportAssignmentType] # raw value is int or str; converted above verbose_logger.debug("GigaChat access token obtained successfully") return access_token, expires_at diff --git a/litellm/llms/gigachat/chat/__init__.py b/litellm/llms/gigachat/chat/__init__.py index eb9492b90b3..0f9be19fedd 100644 --- a/litellm/llms/gigachat/chat/__init__.py +++ b/litellm/llms/gigachat/chat/__init__.py @@ -5,8 +5,8 @@ GigaChat Chat Module from .streaming import GigaChatModelResponseIterator from .transformation import GigaChatConfig, GigaChatError -__all__ = [ +__all__ = ( "GigaChatConfig", "GigaChatError", "GigaChatModelResponseIterator", -] +) diff --git a/litellm/llms/gigachat/chat/streaming.py b/litellm/llms/gigachat/chat/streaming.py index 1b207057537..338d37d810a 100644 --- a/litellm/llms/gigachat/chat/streaming.py +++ b/litellm/llms/gigachat/chat/streaming.py @@ -4,6 +4,7 @@ GigaChat Streaming Response Handler import json import uuid +from collections.abc import Mapping, Sequence from typing import Any, Final from litellm.llms.gigachat.utils import convert_usage @@ -27,14 +28,9 @@ class GigaChatModelResponseIterator: self.response_iterator = self.streaming_response self.json_mode = json_mode - def chunk_parser(self, chunk: dict) -> GenericStreamingChunk: + def chunk_parser(self, chunk: Mapping[str, object]) -> GenericStreamingChunk: """Parse a single streaming chunk from GigaChat.""" - text = "" - tool_use: ChatCompletionToolCallChunk | None = None - is_finished = False - finish_reason: str | None = None - - choices: Final = chunk.get("choices", []) + choices: Sequence = chunk.get("choices") or () # mutable-ok: tuple literal as default if not choices: return GenericStreamingChunk( text="", @@ -46,36 +42,41 @@ class GigaChatModelResponseIterator: ) choice: Final = choices[0] - delta: Final = choice.get("delta", {}) - finish_reason = choice.get("finish_reason") + delta: Mapping[str, object] = choice.get("delta") or {} # mutable-ok: empty dict default for get + chunk_finish_reason: Final = choice.get("finish_reason") # Extract text content - text = delta.get("content", "") or "" + text: Final = delta.get("content", "") or "" + + usage_block: ChatCompletionUsageBlock | None = None + tool_use: ChatCompletionToolCallChunk | None = None + finish_reason: str | None = chunk_finish_reason # Handle function_call in stream - if finish_reason == "function_call" and delta.get("function_call"): + if chunk_finish_reason == "function_call" and delta.get("function_call"): func_call: Final = delta["function_call"] - args = func_call.get("arguments", {}) - - if isinstance(args, dict): - args = json.dumps(args, ensure_ascii=False) + args_raw: Final = func_call.get("arguments") or {} + args_str: str + if isinstance(args_raw, dict): + args_str = json.dumps(args_raw, ensure_ascii=False) # rebind-ok: build from dict + else: + args_str = str(args_raw) tool_use = ChatCompletionToolCallChunk( id=f"call_{uuid.uuid4().hex[:24]}", type="function", function=ChatCompletionToolCallFunctionChunk( name=func_call.get("name", ""), - arguments=args, + arguments=args_str, ), index=0, ) finish_reason = "tool_calls" - usage_block = None - if finish_reason == "stop": - usage_data = chunk.get("usage", {}) + if chunk_finish_reason == "stop": + usage_data: Final = chunk.get("usage") or {} # mutable-ok: empty dict default if usage_data: - usage = convert_usage(usage_data) + usage: Final = convert_usage(usage_data) usage_block = ChatCompletionUsageBlock( prompt_tokens=usage.prompt_tokens, completion_tokens=usage.completion_tokens, @@ -88,13 +89,10 @@ class GigaChatModelResponseIterator: ), ) - if finish_reason is not None: - is_finished = True - return GenericStreamingChunk( text=text, tool_use=tool_use, - is_finished=is_finished, + is_finished=chunk_finish_reason is not None, finish_reason=finish_reason or "", usage=usage_block, index=choice.get("index", 0), diff --git a/litellm/llms/gigachat/chat/transformation.py b/litellm/llms/gigachat/chat/transformation.py index 920a067eaa9..b625e9722f8 100644 --- a/litellm/llms/gigachat/chat/transformation.py +++ b/litellm/llms/gigachat/chat/transformation.py @@ -9,7 +9,7 @@ from __future__ import annotations import json import time import uuid -from collections.abc import AsyncIterator, Iterator +from collections.abc import AsyncIterator, Iterator, Mapping, Sequence from typing import TYPE_CHECKING, Any, Final import httpx @@ -88,8 +88,8 @@ class GigaChatConfig(BaseConfig): api_base: str | None, api_key: str | None, model: str, - optional_params: dict, - litellm_params: dict, + optional_params: Mapping[str, object], + litellm_params: Mapping[str, object], stream: bool | None = None, ) -> str: """Get complete API URL for chat completions.""" @@ -98,14 +98,14 @@ class GigaChatConfig(BaseConfig): def validate_environment( self, - headers: dict, + headers: dict, # mutable-ok: mutates in place per GigaChat OAuth setup model: str, - messages: list[AllMessageValues], - optional_params: dict, - litellm_params: dict, + messages: Sequence[AllMessageValues], + optional_params: Mapping[str, object], + litellm_params: Mapping[str, object], api_key: str | None = None, api_base: str | None = None, - ) -> dict: + ) -> dict: # mutable-ok: base class contract returns dict for httpx """ Set up headers with OAuth token. """ @@ -123,9 +123,9 @@ class GigaChatConfig(BaseConfig): return headers - def get_supported_openai_params(self, model: str) -> list[str]: + def get_supported_openai_params(self, model: str) -> list[str]: # mutable-ok: base class contract returns list """Return list of supported OpenAI parameters.""" - return [ + return [ # mutable-ok: base class contract returns list "stream", "temperature", "top_p", @@ -141,11 +141,11 @@ class GigaChatConfig(BaseConfig): def map_openai_params( self, - non_default_params: dict, - optional_params: dict, + non_default_params: Mapping[str, object], + optional_params: dict, # mutable-ok: mutated in place per GigaChat mapping model: str, drop_params: bool, - ) -> dict: + ) -> dict: # mutable-ok: base class contract returns dict """Map OpenAI parameters to GigaChat parameters.""" for param, value in non_default_params.items(): if param == "stream": @@ -182,25 +182,25 @@ class GigaChatConfig(BaseConfig): schema_name = json_schema.get("name", "structured_output") schema = json_schema.get("schema", {}) - function_def = { + function_def = { # mutable-ok: request payload for httpx "name": schema_name, "description": f"Output structured response: {schema_name}", "parameters": schema, } if "functions" not in optional_params: - optional_params["functions"] = [] + optional_params["functions"] = [] # mutable-ok: list for httpx optional_params["functions"].append(function_def) - optional_params["function_call"] = {"name": schema_name} + optional_params["function_call"] = {"name": schema_name} # mutable-ok: request payload optional_params["_structured_output"] = True return optional_params - def _convert_tools_to_functions(self, tools: list[dict]) -> list[dict]: + def _convert_tools_to_functions(self, tools: Sequence) -> Sequence[dict]: """Convert OpenAI tools format to GigaChat functions format.""" - functions: Final = [] + functions: Final[list[dict]] = [] # mutable-ok: accumulator for building functions list for tool in tools: - if tool.get("type") == "function": + if isinstance(tool, dict) and tool.get("type") == "function": func = tool.get("function", {}) functions.append( { @@ -211,7 +211,7 @@ class GigaChatConfig(BaseConfig): ) return functions - def _map_tool_choice(self, tool_choice: str | dict) -> str | dict | None: + def _map_tool_choice(self, tool_choice: str | Mapping[str, object]) -> str | Mapping[str, object] | None: """ Map OpenAI tool_choice to GigaChat function_call format. @@ -271,7 +271,7 @@ class GigaChatConfig(BaseConfig): verbose_logger.error("Failed to upload image: %s", e) return None - def _transform_list_content(self, content: list) -> tuple[str, list[str]]: + def _transform_list_content(self, content: Sequence) -> tuple[str, Sequence[str]]: """ Extract text and image attachments from a multimodal message content list. @@ -281,8 +281,8 @@ class GigaChatConfig(BaseConfig): Returns: Tuple of (combined text, list of attachment file ids) """ - texts = [] - attachments = [] + texts: Final[list[str]] = [] # mutable-ok: accumulator + attachments: Final[list[str]] = [] # mutable-ok: accumulator for part in content: if isinstance(part, dict): if part.get("type") == "text": @@ -291,24 +291,24 @@ class GigaChatConfig(BaseConfig): # Extract image URL and upload to GigaChat image_url = part.get("image_url", {}) if isinstance(image_url, str): - url = image_url + url: Final = image_url else: - url = image_url.get("url", "") + url: Final = image_url.get("url", "") if url: - file_id = self._upload_image(url) + file_id = self._upload_image(url) # rebind-ok: inside for loop, no outer binding if file_id: attachments.append(file_id) - text = "\n".join(texts) if texts else "" + text: Final = "\n".join(texts) if texts else "" return text, attachments def transform_request( self, model: str, - messages: list[AllMessageValues], - optional_params: dict, - litellm_params: dict, - headers: dict, - ) -> dict: + messages: Sequence[AllMessageValues], + optional_params: Mapping[str, object], + litellm_params: Mapping[str, object], + headers: Mapping[str, object], + ) -> dict: # mutable-ok: request payload sent to httpx """Transform OpenAI request to GigaChat format.""" # Transform messages giga_messages: Final = self._transform_messages(messages) @@ -339,9 +339,9 @@ class GigaChatConfig(BaseConfig): return request_data - def _transform_messages(self, messages: list[AllMessageValues]) -> list[dict]: + def _transform_messages(self, messages: Sequence[AllMessageValues]) -> Sequence[dict]: """Transform OpenAI messages to GigaChat format.""" - transformed: Final = [] + transformed: Final[list[dict]] = [] # mutable-ok: accumulator for building transformed messages for i, msg in enumerate(messages): message = dict(msg) @@ -400,10 +400,10 @@ class GigaChatConfig(BaseConfig): raw_response: httpx.Response, model_response: ModelResponse, logging_obj: LiteLLMLoggingObj, - request_data: dict, - messages: list[AllMessageValues], - optional_params: dict, - litellm_params: dict, + request_data: Mapping[str, object], + messages: Sequence[AllMessageValues], + optional_params: Mapping[str, object], + litellm_params: Mapping[str, object], encoding: Any, api_key: str | None = None, json_mode: bool | None = None, @@ -419,7 +419,7 @@ class GigaChatConfig(BaseConfig): is_structured_output: Final = optional_params.get("_structured_output", False) - choices: Final = [] + choices: Final[list[Choices]] = [] # mutable-ok: accumulator for building response choices for choice in response_json.get("choices", []): message_data = choice.get("message", {}) finish_reason = choice.get("finish_reason", "stop") diff --git a/litellm/llms/gigachat/file_handler.py b/litellm/llms/gigachat/file_handler.py index 6dfd4d74be7..163e944f124 100644 --- a/litellm/llms/gigachat/file_handler.py +++ b/litellm/llms/gigachat/file_handler.py @@ -9,6 +9,7 @@ import base64 import hashlib import re import uuid +from collections.abc import Mapping from typing import Final from litellm._logging import verbose_logger @@ -80,7 +81,7 @@ def upload_file_sync( image_url: str, credentials: str | None = None, api_base: str | None = None, - litellm_params: dict | None = None, + litellm_params: Mapping[str, object] | None = None, ) -> str | None: """ Upload file to GigaChat and return file_id (sync). @@ -146,7 +147,7 @@ async def upload_file_async( image_url: str, credentials: str | None = None, api_base: str | None = None, - litellm_params: dict | None = None, + litellm_params: Mapping[str, object] | None = None, ) -> str | None: """ Upload file to GigaChat and return file_id (async). diff --git a/litellm/llms/gigachat/passthrough/__init__.py b/litellm/llms/gigachat/passthrough/__init__.py index 9e5f9b8ed77..a66a078dbeb 100644 --- a/litellm/llms/gigachat/passthrough/__init__.py +++ b/litellm/llms/gigachat/passthrough/__init__.py @@ -4,4 +4,4 @@ GigaChat passthrough Module from .transformation import GigaChatPassthroughConfig -__all__ = ["GigaChatPassthroughConfig"] +__all__ = ("GigaChatPassthroughConfig",) diff --git a/litellm/llms/gigachat/passthrough/transformation.py b/litellm/llms/gigachat/passthrough/transformation.py index b619b159c6b..ec441c53cc4 100644 --- a/litellm/llms/gigachat/passthrough/transformation.py +++ b/litellm/llms/gigachat/passthrough/transformation.py @@ -1,7 +1,8 @@ from __future__ import annotations import json -from typing import TYPE_CHECKING +from collections.abc import Mapping, Sequence +from typing import TYPE_CHECKING, Final import httpx @@ -21,7 +22,7 @@ if TYPE_CHECKING: class GigaChatPassthroughConfig(BasePassthroughConfig): - def is_streaming_request(self, endpoint: str, request_data: dict) -> bool: + def is_streaming_request(self, endpoint: str, request_data: Mapping[str, object]) -> bool: return request_data.get("stream", False) def get_complete_url( @@ -30,16 +31,16 @@ class GigaChatPassthroughConfig(BasePassthroughConfig): api_key: str | None, model: str, endpoint: str, - request_query_params: dict | None, - litellm_params: dict, + request_query_params: Mapping[str, object] | None, + litellm_params: Mapping[str, object], ) -> tuple[URL, str]: """Get complete API URL for chat completions.""" - base_target_url = self.get_api_base(api_base) + base_target_url: Final = self.get_api_base(api_base) if base_target_url is None: raise Exception("GigaChat api base not found") - complete_url = f"{base_target_url}/{endpoint.lstrip('/')}" + complete_url: Final = f"{base_target_url}/{endpoint.lstrip('/')}" return ( httpx.URL(complete_url), @@ -48,23 +49,23 @@ class GigaChatPassthroughConfig(BasePassthroughConfig): def validate_environment( self, - headers: dict, + headers: dict, # mutable-ok: mutates in place to set OAuth headers model: str, - messages: list[AllMessageValues], - optional_params: dict, - litellm_params: dict, + messages: Sequence[AllMessageValues], + optional_params: Mapping[str, object], + litellm_params: Mapping[str, object], api_key: str | None = None, api_base: str | None = None, - ) -> dict: + ) -> dict: # mutable-ok: base class contract returns dict for httpx """ Set up headers with OAuth token. """ # Get access token - access_token = get_access_token(credentials=api_key, litellm_params=litellm_params) + access_token: Final = get_access_token(credentials=api_key, litellm_params=litellm_params) - headers["Authorization"] = f"Bearer {access_token}" - headers["Content-Type"] = "application/json" - headers["Accept"] = "application/json" + headers["Authorization"] = f"Bearer {access_token}" # rebind-ok: mutating for OAuth setup + headers["Content-Type"] = "application/json" # rebind-ok: mutating for OAuth setup + headers["Accept"] = "application/json" # rebind-ok: mutating for OAuth setup return headers @@ -73,7 +74,7 @@ class GigaChatPassthroughConfig(BasePassthroughConfig): model: str, custom_llm_provider: str, httpx_response: Response, - request_data: dict, + request_data: Mapping[str, object], logging_obj: LiteLLMLoggingObj, endpoint: str, ) -> CostResponseTypes | None: @@ -83,7 +84,7 @@ class GigaChatPassthroughConfig(BasePassthroughConfig): # cost tracking only for completions and embeddings if "completions" in endpoint: - provider_chat_config = ProviderConfigManager.get_provider_chat_config( + provider_chat_config: Final = ProviderConfigManager.get_provider_chat_config( provider=LlmProviders(custom_llm_provider), model=model, ) @@ -93,12 +94,12 @@ class GigaChatPassthroughConfig(BasePassthroughConfig): litellm_model_response: ModelResponse = provider_chat_config.transform_response( model=model, - messages=request_data.get("messages", []), + messages=request_data.get("messages", []), # mutable-ok: empty list default for transform_response raw_response=httpx_response, model_response=ModelResponse(), logging_obj=logging_obj, - optional_params={}, - litellm_params={}, + optional_params={}, # mutable-ok: empty dict kwarg for transform_response + litellm_params={}, # mutable-ok: empty dict kwarg for transform_response api_key="", request_data=request_data, encoding=encoding, @@ -107,7 +108,7 @@ class GigaChatPassthroughConfig(BasePassthroughConfig): return litellm_model_response if "embeddings" in endpoint: - provider_embedding_config = ProviderConfigManager.get_provider_embedding_config( + provider_embedding_config: Final = ProviderConfigManager.get_provider_embedding_config( provider=LlmProviders(custom_llm_provider), model=model, ) @@ -115,15 +116,17 @@ class GigaChatPassthroughConfig(BasePassthroughConfig): if provider_embedding_config is None: raise ValueError(f"No provider config found for model: {model}") - litellm_embedding_response: EmbeddingResponse = provider_embedding_config.transform_embedding_response( - model=model, - raw_response=httpx_response, - model_response=EmbeddingResponse(), - logging_obj=logging_obj, - optional_params={}, - api_key="", - request_data=request_data, - litellm_params={}, + litellm_embedding_response: Final[EmbeddingResponse] = ( + provider_embedding_config.transform_embedding_response( + model=model, + raw_response=httpx_response, + model_response=EmbeddingResponse(), + logging_obj=logging_obj, + optional_params={}, # mutable-ok: empty dict kwarg for transform_embedding_response + api_key="", + request_data=request_data, + litellm_params={}, # mutable-ok: empty dict kwarg for transform_embedding_response + ) ) return litellm_embedding_response @@ -132,7 +135,7 @@ class GigaChatPassthroughConfig(BasePassthroughConfig): def handle_logging_collected_chunks( self, - all_chunks: list[str], + all_chunks: Sequence[str], litellm_logging_obj: LiteLLMLoggingObj, model: str, custom_llm_provider: str, @@ -151,7 +154,7 @@ class GigaChatPassthroughConfig(BasePassthroughConfig): from litellm.main import stream_chunk_builder from litellm.types.utils import ModelResponseStream - all_translated_chunks = [] + all_translated_chunks: Final[list[object]] = [] # mutable-ok: accumulator for chunk in all_chunks: if isinstance(chunk, bytes): @@ -179,7 +182,7 @@ class GigaChatPassthroughConfig(BasePassthroughConfig): if isinstance(translated_chunk, dict) and generic_chunk_has_all_required_fields(translated_chunk): chunk_obj = convert_generic_chunk_to_model_response_stream( - translated_chunk # type: ignore[arg-type] # validated TypedDict + translated_chunk # pyright: ignore[reportArgumentType] # validated TypedDict ) elif isinstance(translated_chunk, ModelResponseStream): chunk_obj = translated_chunk @@ -209,5 +212,5 @@ class GigaChatPassthroughConfig(BasePassthroughConfig): def get_base_model(model: str) -> str | None: return model - def get_models(self, api_key: str | None = None, api_base: str | None = None) -> list[str]: + def get_models(self, api_key: str | None = None, api_base: str | None = None) -> Sequence[str]: return super().get_models(api_key, api_base) diff --git a/litellm/llms/gigachat/utils.py b/litellm/llms/gigachat/utils.py index d2b18f30b5d..aa30552a7ec 100644 --- a/litellm/llms/gigachat/utils.py +++ b/litellm/llms/gigachat/utils.py @@ -1,28 +1,31 @@ +from collections.abc import Mapping +from typing import Final + from litellm.secret_managers.main import get_secret_str from litellm.types.utils import PromptTokensDetailsWrapper, Usage # GigaChat API endpoint -GIGACHAT_BASE_URL = "https://gigachat.devices.sberbank.ru/api/v1" +GIGACHAT_BASE_URL: Final = "https://gigachat.devices.sberbank.ru/api/v1" -def convert_usage(usage_data: dict[str, int]) -> Usage: - prompt_tokens = usage_data.get("prompt_tokens", 0) - completion_tokens = usage_data.get("completion_tokens", 0) - precached_prompt_tokens = usage_data.get("precached_prompt_tokens", 0) - total_tokens = usage_data.get("total_tokens", 0) +def convert_usage(usage_data: Mapping[str, int]) -> Usage: + prompt_tokens: Final = usage_data.get("prompt_tokens", 0) + completion_tokens: Final = usage_data.get("completion_tokens", 0) + precached_prompt_tokens: Final = usage_data.get("precached_prompt_tokens", 0) + total_tokens: Final = usage_data.get("total_tokens", 0) - prompt_tokens += precached_prompt_tokens - total_tokens += precached_prompt_tokens + prompt_tokens_total: Final = prompt_tokens + precached_prompt_tokens + total_tokens_total: Final = total_tokens + precached_prompt_tokens - prompt_tokens_details = None + prompt_tokens_details: PromptTokensDetailsWrapper | None = None if precached_prompt_tokens > 0: prompt_tokens_details = PromptTokensDetailsWrapper(cached_tokens=precached_prompt_tokens) return Usage( - prompt_tokens=prompt_tokens, + prompt_tokens=prompt_tokens_total, completion_tokens=completion_tokens, prompt_tokens_details=prompt_tokens_details, - total_tokens=total_tokens, + total_tokens=total_tokens_total, ) diff --git a/litellm/passthrough/main.py b/litellm/passthrough/main.py index f0c1a802827..9d2ca66965b 100644 --- a/litellm/passthrough/main.py +++ b/litellm/passthrough/main.py @@ -50,9 +50,9 @@ class AsyncPassthroughStreamingResponse(AsyncGenerator[Any, Any]): self._iterator: AsyncGenerator[bytes, Any] self._litellm_logging_obj = litellm_logging_obj self._provider_config = provider_config - self._raw_bytes: list[bytes] = [] + self._raw_bytes: list[bytes] = [] # mutable-ok: instance buffer for streaming chunks self._flush_scheduled = False - self._background_tasks: set[asyncio.Task] = set() + self._background_tasks: set[asyncio.Task] = set() # mutable-ok: instance set for background task tracking @property def status_code(self) -> int: @@ -176,7 +176,7 @@ class PassthroughStreamingResponse(Generator[Any, Any, Any]): self._litellm_logging_obj = litellm_logging_obj self._provider_config = provider_config self._iterator: Generator[bytes, Any, Any] = _as_generator(response.iter_bytes()) - self._raw_bytes: list[bytes] = [] + self._raw_bytes: list[bytes] = [] # mutable-ok: instance buffer for streaming chunks self._flush_scheduled = False def _start_flush(self) -> None: diff --git a/litellm/proxy/common_request_processing.py b/litellm/proxy/common_request_processing.py index 3a53fb275d5..ca94d6dc292 100644 --- a/litellm/proxy/common_request_processing.py +++ b/litellm/proxy/common_request_processing.py @@ -1434,7 +1434,7 @@ class ProxyBaseLLMRequestProcessing: Proxy/custom headers win on key collisions. """ - excluded_headers = { + excluded_headers = { # mutable-ok: set of header names to exclude from forwarding "transfer-encoding", "content-encoding", "set-cookie", @@ -1447,7 +1447,7 @@ class ProxyBaseLLMRequestProcessing: "upgrade", } - merged_headers = { + merged_headers = { # mutable-ok: dict comprehension for merged headers forwarded to httpx key: value for key, value in dict(response_headers or {}).items() if key.lower() not in excluded_headers } merged_headers.update(custom_headers) @@ -2448,7 +2448,7 @@ class ProxyBaseLLMRequestProcessing: # For passthrough routes, stream directly without error parsing # since we're dealing with raw binary data (e.g., AWS event streams) return StreamingResponse( - content=generator, # type: ignore[arg-type] + content=generator, # pyright: ignore[reportArgumentType] # generator-configured StreamingResponse status_code=getattr(response, "status_code", status.HTTP_200_OK), media_type=self._passthrough_event_stream_media_type(), headers=streaming_headers, diff --git a/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py b/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py index 74ce4c751e3..b7a1c71a9e5 100644 --- a/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py @@ -2665,8 +2665,8 @@ def create_generic_websocket_passthrough_endpoint( @router.api_route( "/gigachat/{endpoint:path}", - methods=["GET", "POST", "PUT", "DELETE", "PATCH"], - tags=["Gigachat Pass-through", "pass-through"], + methods=["GET", "POST", "PUT", "DELETE", "PATCH"], # mutable-ok: FastAPI route methods + tags=["Gigachat Pass-through", "pass-through"], # mutable-ok: FastAPI route tags ) async def gigachat_proxy_route( endpoint: str, @@ -2700,7 +2700,9 @@ async def gigachat_proxy_route( if model: is_router_model = is_passthrough_request_using_router_model(request_body, llm_router) elif any(word in endpoint for word in ("completions", "embeddings")): - raise HTTPException(status_code=400, detail={"error": "Model is required in request body"}) + raise HTTPException( + status_code=400, detail={"error": "Model is required in request body"} + ) # mutable-ok: HTTPException detail dict # If router model, use dedicated router passthrough handler # This uses the same common processing path as non-router models @@ -2730,16 +2732,16 @@ async def gigachat_proxy_route( "Gigachat passthrough: Using direct Gigachat model '%s' for endpoint '%s'", model, endpoint ) - data: Dict[str, Any] = {} + data: Dict[str, Any] = {} # mutable-ok: request body mutated in place by proxy pipeline data["method"] = request.method data["endpoint"] = endpoint data["json"] = request_body data["custom_llm_provider"] = "gigachat" - client = get_async_httpx_client( # type: ignore + client = get_async_httpx_client( llm_provider=LlmProviders.GIGACHAT, - params={ + params={ # mutable-ok: httpx client params "timeout": httpx.Timeout(timeout=600.0, connect=5.0), "ssl_verify": False, }, @@ -2823,7 +2825,7 @@ async def handle_gigachat_passthrough_router_model( data: Dict[str, Any] = await _read_request_body(request=request) if user_api_key_dict is not None: if data.get("metadata") is None: - data["metadata"] = {} + data["metadata"] = {} # mutable-ok: metadata dict mutated in place if hasattr(user_api_key_dict, "user_id") and user_api_key_dict.user_id is not None: data["metadata"]["user_api_key_user_id"] = user_api_key_dict.user_id if hasattr(user_api_key_dict, "team_id") and user_api_key_dict.team_id is not None: @@ -2847,7 +2849,7 @@ async def handle_gigachat_passthrough_router_model( data["custom_llm_provider"] = "gigachat" # Remove sensitive keys from data - keys = [ + keys = [ # mutable-ok: list of keys to remove from data "gigachat_auth_url", "gigachat_access_token", "gigachat_scope", @@ -2857,9 +2859,9 @@ async def handle_gigachat_passthrough_router_model( for key in keys: data.pop(key, None) - client = get_async_httpx_client( # type: ignore + client = get_async_httpx_client( llm_provider=LlmProviders.GIGACHAT, - params={ + params={ # mutable-ok: httpx client params "timeout": httpx.Timeout(timeout=600.0, connect=5.0), "ssl_verify": False, },