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

fix(lint): Partly fix basedpyright lint issues

Browse source
This commit is contained in:
KnyazSh 2026-08-23 15:27:55 +00:00
parent 0fc3bccef8
commit ef6cb77f91
9 changed files with 109 additions and 119 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -17,7 +17,6 @@ from litellm.caching.caching import InMemoryCache
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.custom_httpx.http_handler import (
HTTPHandler,
_get_httpx_client,
get_async_httpx_client,
)
from litellm.secret_managers.main import get_secret_str
@ -57,7 +56,7 @@ def _get_scope() -> str:
def _get_http_client() -> HTTPHandler:
"""Get cached httpx client with SSL verification disabled."""
return _get_httpx_client(params={"ssl_verify": False})
return HTTPHandler(ssl_verify=False)
def get_access_token(
@ -101,24 +100,22 @@ def get_access_token(
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
_token, _expires_at = cached
# Check if token is still valid (with buffer)
if time.time() * 1000 < expires_at - TOKEN_EXPIRY_BUFFER_MS:
if time.time() * 1000 < _expires_at - TOKEN_EXPIRY_BUFFER_MS:
verbose_logger.debug("Using cached GigaChat access token")
return token
return _token
# Request new token
token, expires_at = _request_token_sync(effective_credentials, effective_scope, effective_auth_url)
new_token, new_expires_at = _request_token_sync(effective_credentials, effective_scope, effective_auth_url)
if expires_at:
if new_expires_at:
# Cache token
ttl_seconds: Final = max(0, (expires_at - TOKEN_EXPIRY_BUFFER_MS - time.time() * 1000) / 1000)
ttl_seconds: Final = max(0, (new_expires_at - TOKEN_EXPIRY_BUFFER_MS - time.time() * 1000) / 1000)
if ttl_seconds > 0:
_token_cache.set_cache(cache_key, (token, expires_at), ttl=ttl_seconds)
_token_cache.set_cache(cache_key, (new_token, new_expires_at), ttl=ttl_seconds)
return token
return new_token
async def get_access_token_async(
@ -149,23 +146,21 @@ async def get_access_token_async(
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:
_token, _expires_at = cached
if time.time() * 1000 < _expires_at - TOKEN_EXPIRY_BUFFER_MS:
verbose_logger.debug("Using cached GigaChat access token")
return token
return _token
# Request new token
token, expires_at = await _request_token_async(effective_credentials, effective_scope, effective_auth_url)
new_token, new_expires_at = await _request_token_async(effective_credentials, effective_scope, effective_auth_url)
if expires_at:
if new_expires_at:
# Cache token
ttl_seconds: Final = max(0, (expires_at - TOKEN_EXPIRY_BUFFER_MS - time.time() * 1000) / 1000)
ttl_seconds: Final = max(0, (new_expires_at - TOKEN_EXPIRY_BUFFER_MS - time.time() * 1000) / 1000)
if ttl_seconds > 0:
_token_cache.set_cache(cache_key, (token, expires_at), ttl=ttl_seconds)
_token_cache.set_cache(cache_key, (new_token, new_expires_at), ttl=ttl_seconds)
return token
return new_token
def _request_token_sync(

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

@ -289,13 +289,14 @@ class GigaChatConfig(BaseConfig):
texts.append(part.get("text", ""))
elif part.get("type") == "image_url":
# Extract image URL and upload to GigaChat
image_url = part.get("image_url", {})
image_url: object = part.get("image_url", {})
upload_url: str
if isinstance(image_url, str):
url: Final = image_url
upload_url = image_url
else:
url: Final = image_url.get("url", "")
if url:
file_id = self._upload_image(url) # rebind-ok: inside for loop, no outer binding
upload_url = str(image_url.get("url", "")) if isinstance(image_url, dict) else ""
if upload_url:
file_id = self._upload_image(upload_url)
if file_id:
attachments.append(file_id)
text: Final = "\n".join(texts) if texts else ""

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

@ -115,10 +115,8 @@ class GigaChatEmbeddingConfig(BaseEmbeddingConfig):
# Normalize input to list
if isinstance(input, str):
input_list: list = [input]
elif isinstance(input, list):
input_list = input
else:
input_list = [input]
input_list = input
# Remove gigachat/ prefix from model if present
model = model.removeprefix("gigachat/")

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

@ -14,7 +14,7 @@ from typing import Final
from litellm._logging import verbose_logger
from litellm.llms.custom_httpx.http_handler import (
_get_httpx_client,
HTTPHandler,
get_async_httpx_client,
)
from litellm.llms.gigachat.utils import get_api_base
@ -52,7 +52,7 @@ def _parse_data_url(data_url: str) -> tuple[bytes, str, str] | None:
def _download_image_sync(url: str) -> tuple[bytes, str, str]:
"""Download image from URL synchronously."""
client: Final = _get_httpx_client(params={"ssl_verify": False})
client: Final = HTTPHandler(ssl_verify=False)
response: Final = client.get(url)
response.raise_for_status()
@ -120,7 +120,7 @@ def upload_file_sync(
base_url: Final = get_api_base(api_base)
upload_url: Final = f"{base_url}/files"
client: Final = _get_httpx_client(params={"ssl_verify": False})
client: Final = HTTPHandler(ssl_verify=False)
response: Final = client.post(
upload_url,
headers={"Authorization": f"Bearer {access_token}"},

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

@ -157,21 +157,13 @@ class GigaChatPassthroughConfig(BasePassthroughConfig):
all_translated_chunks: Final[list[object]] = [] # mutable-ok: accumulator
for chunk in all_chunks:
if isinstance(chunk, bytes):
chunk = chunk.decode("utf-8", errors="ignore")
if isinstance(chunk, str):
chunk = chunk.strip()
if not chunk or chunk == "[DONE]":
continue
chunk = chunk.removeprefix("data: ")
try:
message = json.loads(chunk)
except json.JSONDecodeError:
continue
elif isinstance(chunk, dict):
message = chunk
else:
chunk = chunk.strip()
if not chunk or chunk == "[DONE]":
continue
chunk = chunk.removeprefix("data: ")
try:
message = json.loads(chunk)
except json.JSONDecodeError:
continue
gigachat_iterator = GigaChatModelResponseIterator(
@ -180,7 +172,7 @@ class GigaChatPassthroughConfig(BasePassthroughConfig):
)
translated_chunk = gigachat_iterator.chunk_parser(chunk=message)
if isinstance(translated_chunk, dict) and generic_chunk_has_all_required_fields(translated_chunk):
if isinstance(translated_chunk, dict) and generic_chunk_has_all_required_fields(translated_chunk): # pyright: ignore[reportUnnecessaryIsInstance] # runtime guard for patched chunk_parser
chunk_obj = convert_generic_chunk_to_model_response_stream(
translated_chunk # pyright: ignore[reportArgumentType] # validated TypedDict
)
@ -212,5 +204,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) -> Sequence[str]:
return super().get_models(api_key, api_base)
def get_models(self, api_key: str | None = None, api_base: str | None = None) -> list[str]:
return list(super().get_models(api_key, api_base))

25
litellm/passthrough/main.py
View file

@ -8,6 +8,7 @@ import asyncio
import contextvars
from collections.abc import AsyncGenerator, AsyncIterator, Coroutine, Generator, Iterator
from functools import partial
from types import TracebackType
from typing import Any, Final, cast
import httpx
@ -124,7 +125,7 @@ class AsyncPassthroughStreamingResponse(AsyncGenerator[Any, Any]):
async def __anext__(self) -> bytes:
if not self._initialized:
await self
await self # pyright: ignore[reportGeneralTypeIssues] # structural type check misses __await__
try:
chunk = await anext(self._iterator)
self._raw_bytes.append(chunk)
@ -140,17 +141,17 @@ class AsyncPassthroughStreamingResponse(AsyncGenerator[Any, Any]):
async def asend(self, value: bytes) -> bytes:
if not self._initialized:
await self
await self # pyright: ignore[reportGeneralTypeIssues] # structural type check misses __await__
return await self._iterator.asend(value)
async def athrow(
self,
typ: type[BaseException],
val: BaseException | None = None,
tb: type | None = None,
typ: BaseException | type[BaseException],
val: BaseException | object = None,
tb: TracebackType | None = None,
) -> bytes:
if not self._initialized:
await self
await self # pyright: ignore[reportGeneralTypeIssues] # structural type check misses __await__
return await self._iterator.athrow(typ, val, tb)
async def aclose(self) -> None:
@ -221,9 +222,9 @@ class PassthroughStreamingResponse(Generator[Any, Any, Any]):
def throw(
self,
typ: type[BaseException],
val: BaseException | None = None,
tb: type | None = None,
typ: BaseException | type[BaseException],
val: BaseException | object = None,
tb: TracebackType | None = None,
) -> bytes:
return self._iterator.throw(typ, val, tb)
@ -552,8 +553,8 @@ def llm_passthrough_route(
else:
return response
except Exception as e:
if provider_config is None:
raise e
# provider_config is guaranteed non-None here due to the earlier guard
assert provider_config is not None
raise base_llm_http_handler._handle_error(
e=e,
provider_config=provider_config,
@ -577,7 +578,7 @@ async def _async_passthrough_request(
# Check if it's a coroutine and await it
if asyncio.iscoroutine(response_result):
if is_streaming_request:
return await AsyncPassthroughStreamingResponse(
return await AsyncPassthroughStreamingResponse( # pyright: ignore[reportGeneralTypeIssues] # structural type check misses __await__
response=response_result,
litellm_logging_obj=litellm_logging_obj,
provider_config=provider_config,

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

@ -47,6 +47,7 @@ from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
create_websocket_passthrough_route,
websocket_passthrough_request,
)
from litellm.proxy.utils import ProxyLogging as ProxyLoggingType
from litellm.proxy.utils import is_known_model
from litellm.proxy.vector_store_endpoints.utils import (
assert_proxy_admin_for_vector_store_index_management,
@ -1676,7 +1677,7 @@ def get_vertex_ai_allowed_incoming_headers(request: Request) -> dict:
def get_vertex_pass_through_handler(
call_type: Literal[discovery, aiplatform],
call_type: Literal["discovery", "aiplatform"], # noqa: UP037
) -> BaseVertexAIPassThroughHandler:
if call_type == "discovery":
return VertexAIDiscoveryPassThroughHandler()
@ -2732,7 +2733,7 @@ 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: dict[str, Any] = {} # mutable-ok: request body mutated in place by proxy pipeline
data["method"] = request.method
data["endpoint"] = endpoint
@ -2784,7 +2785,7 @@ async def handle_gigachat_passthrough_router_model(
fastapi_response: Response,
llm_router: litellm.Router,
user_api_key_dict: UserAPIKeyAuth,
proxy_logging_obj: ProxyLogging,
proxy_logging_obj: ProxyLoggingType,
general_settings: dict,
proxy_config: ProxyConfig,
select_data_generator: Callable,
@ -2822,7 +2823,7 @@ async def handle_gigachat_passthrough_router_model(
# Detect streaming based on request body
is_streaming = request_body.get("stream", False)
data: Dict[str, Any] = await _read_request_body(request=request)
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"] = {} # mutable-ok: metadata dict mutated in place

46
tests/test_litellm/llms/gigachat/passthrough/test_gigachat_passthrough_transformation.py
View file

@ -299,13 +299,13 @@ class TestGigaChatPassthroughConfig:
assert result.choices[0].message.content == "Hello world"
def test_handle_logging_collected_chunks_with_bytes_chunks(self):
"""Test converting bytes chunks to model response."""
"""Test converting string chunks to model response (bytes pre-decoded upstream)."""
config = GigaChatPassthroughConfig()
logging_obj = MagicMock()
chunks = [
b'{"choices": [{"delta": {"content": "Hi"}, "index": 0}]}',
b'{"choices": [{"delta": {}, "finish_reason": "stop", "index": 0}], "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}}',
'{"choices": [{"delta": {"content": "Hi"}, "index": 0}]}',
'{"choices": [{"delta": {}, "finish_reason": "stop", "index": 0}], "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}}',
]
result = config.handle_logging_collected_chunks(
@ -343,26 +343,28 @@ class TestGigaChatPassthroughConfig:
assert result.choices[0].message.content == "test"
def test_handle_logging_collected_chunks_with_dict_chunks(self):
"""Test converting dict chunks directly."""
"""Test converting string-serialized dict chunks (dicts pre-serialized upstream)."""
config = GigaChatPassthroughConfig()
logging_obj = MagicMock()
chunks = [
{"choices": [{"delta": {"content": "direct"}, "index": 0}]},
{
"choices": [
{
"delta": {},
"finish_reason": "stop",
"index": 0,
}
],
"usage": {
"prompt_tokens": 1,
"completion_tokens": 1,
"total_tokens": 2,
},
},
'{"choices": [{"delta": {"content": "direct"}, "index": 0}]}',
json.dumps(
{
"choices": [
{
"delta": {},
"finish_reason": "stop",
"index": 0,
}
],
"usage": {
"prompt_tokens": 1,
"completion_tokens": 1,
"total_tokens": 2,
},
}
),
]
result = config.handle_logging_collected_chunks(
@ -587,12 +589,12 @@ class TestGigaChatPassthroughConfig:
assert result is None
def test_handle_logging_collected_chunks_skips_unsupported_chunk_type(self):
"""Test that unsupported chunk types (int, float, etc.) are skipped."""
"""Test that unsupported chunk types (non-JSON str) are skipped."""
config = GigaChatPassthroughConfig()
logging_obj = MagicMock()
# The chunk is an int which doesn't match str/bytes/dict
chunks: list = [42, "not-a-real-chunk"]
# Both are valid str chunks; "not-a-valid-json" fails json.loads, int is not a str
chunks: list[str] = ["not-a-valid-json"]
result = config.handle_logging_collected_chunks(
all_chunks=chunks,

60
tests/test_litellm/llms/gigachat/test_file_handler.py
View file

@ -128,14 +128,14 @@ class TestParseDataUrl:
class TestDownloadImageSync:
@patch(f"{FILE_MODULE}._get_httpx_client")
def test_downloads_image_successfully(self, mock_get_client):
@patch(f"{FILE_MODULE}.HTTPHandler")
def test_downloads_image_successfully(self, mock_http_handler_cls):
mock_client = MagicMock()
mock_response = MagicMock()
mock_response.content = b"fake-image-bytes"
mock_response.headers = {"content-type": "image/jpeg"}
mock_client.get.return_value = mock_response
mock_get_client.return_value = mock_client
mock_http_handler_cls.return_value = mock_client
content_bytes, content_type, ext = file_handler._download_image_sync("https://example.com/img.jpg")
@ -144,41 +144,41 @@ class TestDownloadImageSync:
assert ext == "jpeg"
mock_client.get.assert_called_once_with("https://example.com/img.jpg")
@patch(f"{FILE_MODULE}._get_httpx_client")
def test_raises_on_http_error(self, mock_get_client):
@patch(f"{FILE_MODULE}.HTTPHandler")
def test_raises_on_http_error(self, mock_http_handler_cls):
mock_client = MagicMock()
mock_client.get.side_effect = httpx.HTTPStatusError(
"Not Found",
request=httpx.Request("GET", "https://example.com/404"),
response=httpx.Response(status_code=404, request=httpx.Request("GET", "https://example.com/404")),
)
mock_get_client.return_value = mock_client
mock_http_handler_cls.return_value = mock_client
with pytest.raises(httpx.HTTPStatusError):
file_handler._download_image_sync("https://example.com/404")
@patch(f"{FILE_MODULE}._get_httpx_client")
def test_parse_content_type_fallback(self, mock_get_client):
@patch(f"{FILE_MODULE}.HTTPHandler")
def test_parse_content_type_fallback(self, mock_http_handler_cls):
mock_client = MagicMock()
mock_response = MagicMock()
mock_response.content = b"data"
mock_response.headers = {}
mock_client.get.return_value = mock_response
mock_get_client.return_value = mock_client
mock_http_handler_cls.return_value = mock_client
_, content_type, ext = file_handler._download_image_sync("https://example.com/img")
assert content_type == "image/jpeg"
assert ext == "jpeg"
@patch(f"{FILE_MODULE}._get_httpx_client")
def test_extracts_extension_from_parametrized_type(self, mock_get_client):
@patch(f"{FILE_MODULE}.HTTPHandler")
def test_extracts_extension_from_parametrized_type(self, mock_http_handler_cls):
mock_client = MagicMock()
mock_response = MagicMock()
mock_response.content = b"data"
mock_response.headers = {"content-type": "image/png; charset=utf-8"}
mock_client.get.return_value = mock_response
mock_get_client.return_value = mock_client
mock_http_handler_cls.return_value = mock_client
_, _, ext = file_handler._download_image_sync("https://example.com/img.png")
@ -235,16 +235,16 @@ class TestDownloadImageAsync:
class TestUploadFileSync:
@patch(f"{FILE_MODULE}.get_api_base", return_value="https://api.example.com")
@patch(f"{FILE_MODULE}.get_access_token", return_value="test-token")
@patch(f"{FILE_MODULE}._get_httpx_client")
@patch(f"{FILE_MODULE}.HTTPHandler")
def test_uploads_base64_image_and_caches(
self, mock_get_client, mock_get_token, mock_get_api_base
self, mock_http_handler_cls, mock_get_token, mock_get_api_base
):
mock_client = MagicMock()
mock_response = MagicMock()
mock_response.json.return_value = {"id": "file-12345"}
mock_response.raise_for_status = MagicMock()
mock_client.post.return_value = mock_response
mock_get_client.return_value = mock_client
mock_http_handler_cls.return_value = mock_client
result = upload_file_sync(
image_url=_RED_PNG_DATA_URL,
@ -268,9 +268,9 @@ class TestUploadFileSync:
@patch(f"{FILE_MODULE}.get_api_base", return_value="https://api.example.com")
@patch(f"{FILE_MODULE}.get_access_token", return_value="test-token")
@patch(f"{FILE_MODULE}._get_httpx_client")
@patch(f"{FILE_MODULE}.HTTPHandler")
def test_returns_cached_file_id(
self, mock_get_client, mock_get_token, mock_get_api_base
self, mock_http_handler_cls, mock_get_token, mock_get_api_base
):
# Pre-populate the cache
url_hash = _get_url_hash(_RED_PNG_DATA_URL)
@ -280,14 +280,14 @@ class TestUploadFileSync:
assert result == "cached-file-id"
# No upload call was made
mock_get_client.return_value.post.assert_not_called()
mock_http_handler_cls.return_value.post.assert_not_called()
@patch(f"{FILE_MODULE}._get_httpx_client")
@patch(f"{FILE_MODULE}.HTTPHandler")
@patch(f"{FILE_MODULE}.get_access_token", return_value="test-token")
@patch(f"{FILE_MODULE}.get_api_base", return_value="https://api.example.com")
@patch(f"{FILE_MODULE}._download_image_sync")
def test_downloads_and_uploads_url_image(
self, mock_download, mock_get_api_base, mock_get_token, mock_get_client
self, mock_download, mock_get_api_base, mock_get_token, mock_http_handler_cls
):
mock_download.return_value = (b"remote-bytes", "image/png", "png")
mock_client = MagicMock()
@ -295,7 +295,7 @@ class TestUploadFileSync:
mock_response.json.return_value = {"id": "file-remote"}
mock_response.raise_for_status = MagicMock()
mock_client.post.return_value = mock_response
mock_get_client.return_value = mock_client
mock_http_handler_cls.return_value = mock_client
result = upload_file_sync(
image_url="https://example.com/remote.png", credentials="creds"
@ -304,11 +304,11 @@ class TestUploadFileSync:
assert result == "file-remote"
mock_download.assert_called_once_with("https://example.com/remote.png")
@patch(f"{FILE_MODULE}._get_httpx_client")
@patch(f"{FILE_MODULE}.HTTPHandler")
@patch(f"{FILE_MODULE}.get_access_token", return_value="test-token")
@patch(f"{FILE_MODULE}.get_api_base", return_value="https://api.example.com")
def test_returns_none_on_upload_failure(
self, mock_get_api_base, mock_get_token, mock_get_client
self, mock_get_api_base, mock_get_token, mock_http_handler_cls
):
mock_client = MagicMock()
mock_client.post.side_effect = httpx.HTTPStatusError(
@ -316,7 +316,7 @@ class TestUploadFileSync:
request=httpx.Request("POST", "https://api.example.com/files"),
response=httpx.Response(status_code=400, request=httpx.Request("POST", "https://api.example.com/files")),
)
mock_get_client.return_value = mock_client
mock_http_handler_cls.return_value = mock_client
# upload_file_sync catches all exceptions and returns None
result = upload_file_sync(
@ -325,18 +325,18 @@ class TestUploadFileSync:
assert result is None
@patch(f"{FILE_MODULE}._get_httpx_client")
@patch(f"{FILE_MODULE}.HTTPHandler")
@patch(f"{FILE_MODULE}.get_access_token", return_value="test-token")
@patch(f"{FILE_MODULE}.get_api_base", return_value="https://api.example.com")
def test_returns_none_when_response_missing_id(
self, mock_get_api_base, mock_get_token, mock_get_client
self, mock_get_api_base, mock_get_token, mock_http_handler_cls
):
mock_client = MagicMock()
mock_response = MagicMock()
mock_response.json.return_value = {"status": "ok"} # no "id" key
mock_response.raise_for_status = MagicMock()
mock_client.post.return_value = mock_response
mock_get_client.return_value = mock_client
mock_http_handler_cls.return_value = mock_client
result = upload_file_sync(
image_url=_RED_PNG_DATA_URL, credentials="creds"
@ -346,9 +346,9 @@ class TestUploadFileSync:
@patch(f"{FILE_MODULE}.get_api_base", return_value="https://api.example.com")
@patch(f"{FILE_MODULE}.get_access_token", return_value="test-token")
@patch(f"{FILE_MODULE}._get_httpx_client")
@patch(f"{FILE_MODULE}.HTTPHandler")
def test_uploads_without_optional_args(
self, mock_get_client, mock_get_token, mock_get_api_base
self, mock_http_handler_cls, mock_get_token, mock_get_api_base
):
"""Verify that credentials, api_base, and litellm_params are optional."""
mock_client = MagicMock()
@ -356,7 +356,7 @@ class TestUploadFileSync:
mock_response.json.return_value = {"id": "file-no-args"}
mock_response.raise_for_status = MagicMock()
mock_client.post.return_value = mock_response
mock_get_client.return_value = mock_client
mock_http_handler_cls.return_value = mock_client
result = upload_file_sync(image_url=_RED_PNG_DATA_URL)