scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-07 02:59:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 103

fix(lint): Partly fix LIT issues

Browse source
This commit is contained in:
KnyazSh 2026-08-26 19:28:42 +00:00
parent 736411bd67
commit 5af54d26b8
8 changed files with 42 additions and 36 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -249,12 +249,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
expires_at: int # rebind-ok: conditionally assigned from str or int
if isinstance(expires_at_raw, str):
expires_at = int(expires_at_raw)
expires_at = int(expires_at_raw) # rebind-ok: conditionally assigned from str or int
else:
expires_at = expires_at_raw # pyright: ignore[reportAssignmentType] # raw value is int or str; converted above
expires_at = expires_at_raw # pyright: ignore[reportAssignmentType] # raw value is int or str; converted above; rebind-ok: conditionally assigned from str or int
verbose_logger.debug("GigaChat access token obtained successfully")
return access_token, expires_at

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

@ -48,15 +48,15 @@ class GigaChatModelResponseIterator:
# Extract text content
text: Final = delta.get("content", "") or ""
usage_block: ChatCompletionUsageBlock | None = None
tool_use: ChatCompletionToolCallChunk | None = None
usage_block: ChatCompletionUsageBlock | None = None # rebind-ok: conditionally assigned after stop detection
tool_use: ChatCompletionToolCallChunk | None = None # rebind-ok: conditionally assigned on function_call
finish_reason: str | None = chunk_finish_reason
# Handle function_call in stream
if chunk_finish_reason == "function_call" and delta.get("function_call"):
func_call: Final = delta["function_call"]
args_raw: Final = func_call.get("arguments") or {}
args_str: str
args_str: str # rebind-ok: conditionally assigned from dict or str
if isinstance(args_raw, dict):
args_str = json.dumps(args_raw, ensure_ascii=False) # rebind-ok: build from dict
else:
@ -76,7 +76,7 @@ class GigaChatModelResponseIterator:
if chunk_finish_reason == "stop":
usage_data: Final = chunk.get("usage") or {} # mutable-ok: empty dict default
if usage_data:
usage: Final = convert_usage(usage_data)
usage = convert_usage(usage_data) # rebind-ok: conditional usage assignment
usage_block = ChatCompletionUsageBlock(
prompt_tokens=usage.prompt_tokens,
completion_tokens=usage.completion_tokens,

4
litellm/llms/gigachat/embedding/transformation.py
View file

@ -114,12 +114,12 @@ class GigaChatEmbeddingConfig(BaseEmbeddingConfig):
"""
# Normalize input to list
if isinstance(input, str):
input_list: list = [input]
input_list: list = [input] # rebind-ok: locally scoped conversion
else:
input_list = input
# Remove gigachat/ prefix from model if present
model = model.removeprefix("gigachat/")
model = model.removeprefix("gigachat/") # rebind-ok: parameter reassignment for normalization
return {
"model": model,

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

@ -92,7 +92,7 @@ class GigaChatPassthroughConfig(BasePassthroughConfig):
if provider_chat_config is None:
raise ValueError(f"No provider config found for model: {model}")
litellm_model_response: ModelResponse = provider_chat_config.transform_response(
litellm_model_response: Final = provider_chat_config.transform_response(
model=model,
messages=request_data.get("messages", []), # mutable-ok: empty list default for transform_response
raw_response=httpx_response,

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

@ -17,9 +17,13 @@ def convert_usage(usage_data: Mapping[str, int]) -> Usage:
prompt_tokens_total: Final = prompt_tokens + precached_prompt_tokens
total_tokens_total: Final = total_tokens + precached_prompt_tokens
prompt_tokens_details: PromptTokensDetailsWrapper | None = None
prompt_tokens_details: PromptTokensDetailsWrapper | None = (
None # rebind-ok: conditionally assigned when cached tokens exist
)
if precached_prompt_tokens > 0:
prompt_tokens_details = PromptTokensDetailsWrapper(cached_tokens=precached_prompt_tokens)
prompt_tokens_details = PromptTokensDetailsWrapper(
cached_tokens=precached_prompt_tokens
) # rebind-ok: conditionally assigned when cached tokens exist
return Usage(
prompt_tokens=prompt_tokens_total,

6
litellm/passthrough/main.py
View file

@ -101,7 +101,7 @@ class AsyncPassthroughStreamingResponse(AsyncGenerator[Any, Any]):
self._flush_scheduled = True
try:
task = asyncio.create_task(
task: Final = asyncio.create_task(
self._litellm_logging_obj.async_flush_passthrough_collected_chunks(
raw_bytes=self._raw_bytes,
provider_config=self._provider_config,
@ -127,7 +127,7 @@ class AsyncPassthroughStreamingResponse(AsyncGenerator[Any, Any]):
if not self._initialized:
await self # pyright: ignore[reportGeneralTypeIssues] # structural type check misses __await__
try:
chunk = await anext(self._iterator)
chunk: Final = await anext(self._iterator)
self._raw_bytes.append(chunk)
except Exception: # noqa: BLE001 # Safe catch-all for cleanup logic
self._start_flush()
@ -205,7 +205,7 @@ class PassthroughStreamingResponse(Generator[Any, Any, Any]):
def __next__(self) -> bytes:
try:
chunk = next(self._iterator)
chunk: Final = next(self._iterator)
self._raw_bytes.append(chunk)
except Exception: # noqa: BLE001 # Safe catch-all for cleanup logic
self._start_flush()

8
litellm/proxy/common_request_processing.py
View file

@ -1448,7 +1448,7 @@ class ProxyBaseLLMRequestProcessing:
Proxy/custom headers win on key collisions.
"""
excluded_headers = { # mutable-ok: set of header names to exclude from forwarding
excluded_headers: Final = { # mutable-ok: set of header names to exclude from forwarding
"transfer-encoding",
"content-encoding",
"set-cookie",
@ -1461,7 +1461,7 @@ class ProxyBaseLLMRequestProcessing:
"upgrade",
}
merged_headers = { # mutable-ok: dict comprehension for merged headers forwarded to httpx
merged_headers: Final = { # 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)
@ -2425,9 +2425,9 @@ class ProxyBaseLLMRequestProcessing:
logging_obj._on_deferred_stream_complete = _on_deferred_stream_complete
if route_type == "allm_passthrough_route":
streaming_headers = custom_headers
streaming_headers = custom_headers # rebind-ok: initial assignment before header merge
if hasattr(response, "headers"):
streaming_headers = ProxyBaseLLMRequestProcessing._merge_passthrough_streaming_headers(
streaming_headers = ProxyBaseLLMRequestProcessing._merge_passthrough_streaming_headers( # rebind-ok: merge result replaces initial assignment
response_headers=getattr(response, "headers", None),
custom_headers=custom_headers,
)

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

@ -76,9 +76,9 @@ if TYPE_CHECKING:
from litellm.proxy.proxy_server import ProxyConfig as _ProxyConfig
from litellm.router import Router
ProxyConfig = _ProxyConfig
ProxyConfig = _ProxyConfig # rebind-ok: conditional type alias
else:
ProxyConfig = Any
ProxyConfig = Any # rebind-ok: runtime fallback
vertex_llm_base: Final = VertexBase()
router: Final = APIRouter()
@ -508,8 +508,8 @@ async def milvus_proxy_route(
status_code=400,
detail=f"collectionName must be a string. Got {type(_raw_collection_name).__name__}",
)
collection_name: str | None = _raw_collection_name
extra_headers = {}
collection_name: str | None = _raw_collection_name # rebind-ok: locally scoped conversion
extra_headers: Final = {} # mutable-ok: dict for extra headers
base_target_url: str | None = None
if not collection_name:
raise HTTPException(
@ -2839,12 +2839,14 @@ async def gigachat_proxy_route(
)
## check for streaming
request_body = await get_request_body(request)
is_router_model = False
request_body: Final = await get_request_body(request)
is_router_model = False # rebind-ok: conditionally set to True when model uses router
model = request_body.get("model")
model: Final = request_body.get("model")
if model:
is_router_model = is_passthrough_request_using_router_model(request_body, llm_router)
is_router_model = is_passthrough_request_using_router_model(
request_body, llm_router
) # rebind-ok: conditionally set to True
elif any(word in endpoint for word in ("completions", "embeddings")):
raise HTTPException(
status_code=400, detail={"error": "Model is required in request body"}
@ -2878,14 +2880,14 @@ async def gigachat_proxy_route(
"Gigachat passthrough: Using direct Gigachat model '%s' for endpoint '%s'", model, endpoint
)
data: dict[str, Any] = {} # mutable-ok: request body mutated in place by proxy pipeline
data: Final[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(
client: Final = get_async_httpx_client(
llm_provider=LlmProviders.GIGACHAT,
params={ # mutable-ok: httpx client params
"timeout": httpx.Timeout(timeout=600.0, connect=5.0),
@ -2894,7 +2896,7 @@ async def gigachat_proxy_route(
)
data["client"] = client
base_llm_response_processor = ProxyBaseLLMRequestProcessing(data=data)
base_llm_response_processor: Final = ProxyBaseLLMRequestProcessing(data=data)
try:
return await base_llm_response_processor.base_passthrough_process_llm_request(
@ -2966,9 +2968,9 @@ async def handle_gigachat_passthrough_router_model(
from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
# Detect streaming based on request body
is_streaming = request_body.get("stream", False)
is_streaming: Final = request_body.get("stream", False)
data: dict[str, Any] = await _read_request_body(request=request)
data: Final[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"] = {} # mutable-ok: metadata dict mutated in place
@ -2995,7 +2997,7 @@ async def handle_gigachat_passthrough_router_model(
data["custom_llm_provider"] = "gigachat"
# Remove sensitive keys from data
keys = [ # mutable-ok: list of keys to remove from data
keys: Final = [ # mutable-ok: list of keys to remove from data
"gigachat_auth_url",
"gigachat_access_token",
"gigachat_scope",
@ -3005,7 +3007,7 @@ async def handle_gigachat_passthrough_router_model(
for key in keys:
data.pop(key, None)
client = get_async_httpx_client(
client: Final = get_async_httpx_client(
llm_provider=LlmProviders.GIGACHAT,
params={ # mutable-ok: httpx client params
"timeout": httpx.Timeout(timeout=600.0, connect=5.0),
@ -3014,12 +3016,12 @@ async def handle_gigachat_passthrough_router_model(
)
data["client"] = client
base_llm_response_processor = ProxyBaseLLMRequestProcessing(data=data)
base_llm_response_processor: Final = ProxyBaseLLMRequestProcessing(data=data)
# Use the common passthrough processing to handle metadata and hooks
# This also handles all response formatting (streaming/non-streaming) and exceptions
try:
result = await base_llm_response_processor.base_passthrough_process_llm_request(
result: Final = await base_llm_response_processor.base_passthrough_process_llm_request(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,