scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-09 22:31:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 133

fix(lint): Partly fix LIT issues

Browse source
This commit is contained in:
KnyazSh 2026-08-23 14:04:10 +00:00
parent 069423ce00
commit 950267e3cd
12 changed files with 167 additions and 152 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
litellm/llms/gigachat/__init__.py
View file

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

50
litellm/llms/gigachat/authenticator.py
View file

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

4
litellm/llms/gigachat/chat/__init__.py
View file

@ -5,8 +5,8 @@ GigaChat Chat Module
from .streaming import GigaChatModelResponseIterator
from .transformation import GigaChatConfig, GigaChatError
__all__ = [
__all__ = (
"GigaChatConfig",
"GigaChatError",
"GigaChatModelResponseIterator",
]
)

46
litellm/llms/gigachat/chat/streaming.py
View file

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

78
litellm/llms/gigachat/chat/transformation.py
View file

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

5
litellm/llms/gigachat/file_handler.py
View file

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

2
litellm/llms/gigachat/passthrough/__init__.py
View file

@ -4,4 +4,4 @@ GigaChat passthrough Module
from .transformation import GigaChatPassthroughConfig
__all__ = ["GigaChatPassthroughConfig"]
__all__ = ("GigaChatPassthroughConfig",)

71
litellm/llms/gigachat/passthrough/transformation.py
View file

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

25
litellm/llms/gigachat/utils.py
View file

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

6
litellm/passthrough/main.py
View file

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

6
litellm/proxy/common_request_processing.py
View file

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

22
litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py
View file

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