From 7ebc144c18a9e9fba9500814cd7cd17afccbc433 Mon Sep 17 00:00:00 2001 From: harish876 Date: Thu, 9 Apr 2026 22:14:46 +0000 Subject: [PATCH 01/14] Add file content streaming support for OpenAI and related utilities - Introduced `afile_content_streaming` and `file_content_streaming` functions in `litellm/files/main.py` to handle asynchronous and synchronous file content streaming. - Added `FileContentStreamingResponse` class in `litellm/files/streaming.py` to manage streaming responses with logging capabilities. - Updated OpenAI API integration in `litellm/llms/openai/openai.py` to support new streaming methods. - Enhanced file content retrieval in `litellm/proxy/openai_files_endpoints/files_endpoints.py` to route requests for streaming. - Added unit tests for the new streaming functionality in `tests/test_litellm/llms/openai/test_openai_file_content_streaming.py` and `tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py`. - Refactored type hints and imports for better clarity and organization across modified files. --- litellm/files/main.py | 152 ++++++++++++- litellm/files/streaming.py | 205 ++++++++++++++++++ litellm/llms/openai/openai.py | 59 ++++- .../openai_files_endpoints/files_endpoints.py | 108 ++++++++- litellm/utils.py | 2 + .../test_openai_file_content_streaming.py | 102 +++++++++ tests/test_litellm/proxy/conftest.py | 24 ++ .../test_files_endpoint.py | 56 +++++ 8 files changed, 699 insertions(+), 9 deletions(-) create mode 100644 litellm/files/streaming.py create mode 100644 tests/test_litellm/llms/openai/test_openai_file_content_streaming.py diff --git a/litellm/files/main.py b/litellm/files/main.py index f7c89e0ba3..865b679e04 100644 --- a/litellm/files/main.py +++ b/litellm/files/main.py @@ -10,7 +10,7 @@ import contextvars import time import uuid as uuid_module from functools import partial -from typing import Any, Coroutine, Dict, Literal, Optional, Union, cast +from typing import Any, AsyncIterator, Coroutine, Dict, Iterator, Literal, Optional, Union, cast import httpx @@ -36,6 +36,7 @@ FileContentProvider = Literal[ import litellm from litellm import get_secret_str +from litellm.files.streaming import FileContentStreamingResponse from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.llms.azure.common_utils import get_azure_credentials @@ -55,10 +56,7 @@ from litellm.types.llms.openai import ( OpenAIFileObject, ) from litellm.types.router import * -from litellm.types.utils import ( - OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS, - LlmProviders, -) +from litellm.types.utils import OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS, LlmProviders from litellm.utils import ( ProviderConfigManager, client, @@ -982,3 +980,147 @@ def file_content( return response except Exception as e: raise e + + +@client +async def afile_content_streaming( + file_id: str, + custom_llm_provider: FileContentProvider = "openai", + extra_headers: Optional[Dict[str, str]] = None, + extra_body: Optional[Dict[str, str]] = None, + chunk_size: int = 1024 * 1024, + **kwargs, +) -> Union[Iterator[bytes], AsyncIterator[bytes]]: + """ + Async wrapper for file_content_streaming. + """ + try: + loop = asyncio.get_running_loop() + kwargs["afile_content_streaming"] = True + model = kwargs.pop("model", None) + + # Use a partial function to pass your keyword arguments + func = partial( + file_content_streaming, + file_id, + model, + custom_llm_provider, + extra_headers, + extra_body, + chunk_size, + **kwargs, + ) + + # Add the context to the function + ctx = contextvars.copy_context() + func_with_context = partial(ctx.run, func) + init_response = await loop.run_in_executor(None, func_with_context) + if asyncio.iscoroutine(init_response): + response = await init_response + else: + response = init_response # type: ignore + + return response + except Exception as e: + raise e + + +@client +def file_content_streaming( + file_id: str, + model: Optional[str] = None, + custom_llm_provider: Optional[Union[FileContentProvider, str]] = None, + extra_headers: Optional[Dict[str, str]] = None, + extra_body: Optional[Dict[str, str]] = None, + chunk_size: int = 1024 * 1024, + **kwargs, +) -> Union[Iterator[bytes], AsyncIterator[bytes]]: + """ + Prototype API: Returns a byte iterator for file contents. + + Supports OpenAI-compatible providers and Azure. + """ + try: + optional_params = GenericLiteLLMParams(**kwargs) + litellm_params_dict = get_litellm_params(**kwargs) + client = kwargs.get("client") + logging_obj = cast( + Optional[LiteLLMLoggingObj], kwargs.get("litellm_logging_obj") + ) + + try: + if model is not None: + _, custom_llm_provider, _, _ = get_llm_provider( + model, custom_llm_provider + ) + except Exception: + pass + + timeout = optional_params.timeout or kwargs.get("request_timeout", 600) or 600 + if ( + timeout is not None + and isinstance(timeout, httpx.Timeout) + and supports_httpx_timeout(cast(str, custom_llm_provider)) is False + ): + timeout = timeout.read or 600 + elif timeout is not None and not isinstance(timeout, httpx.Timeout): + timeout = float(timeout) # type: ignore + elif timeout is None: + timeout = 600.0 + + _is_async = kwargs.pop("afile_content_streaming", False) is True + + if logging_obj is not None: + logging_obj.model = model or "" + logging_obj.model_call_details["model"] = model or "" + logging_obj.model_call_details["custom_llm_provider"] = custom_llm_provider + + litellm_params = logging_obj.model_call_details.get("litellm_params", {}) or {} + if optional_params.api_base is not None: + litellm_params["api_base"] = optional_params.api_base + logging_obj.model_call_details["litellm_params"] = litellm_params + + response = cast(Union[Iterator[bytes], AsyncIterator[bytes]], iter(())) + if custom_llm_provider in OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS: + openai_creds = get_openai_credentials( + api_base=optional_params.api_base, + api_key=optional_params.api_key, + organization=optional_params.organization, + ) + response = openai_files_instance.file_content_streaming( + _is_async=_is_async, + file_content_request=FileContentRequest( + file_id=file_id, + extra_headers=extra_headers, + extra_body=extra_body, + ), + api_base=openai_creds.api_base, + api_key=openai_creds.api_key, + timeout=timeout, + max_retries=optional_params.max_retries, + organization=openai_creds.organization, + chunk_size=chunk_size, + ) + else: + raise litellm.exceptions.BadRequestError( + message="LiteLLM doesn't support {} for 'file_content'. Supported providers are 'openai', 'azure', 'vertex_ai', 'bedrock', 'manus', 'anthropic'.".format( + custom_llm_provider + ), + model="n/a", + llm_provider=custom_llm_provider, + response=httpx.Response( + status_code=400, + content="Unsupported provider", + request=httpx.Request(method="create_thread", url="https://github.com/BerriAI/litellm"), # type: ignore + ), + ) + + return FileContentStreamingResponse( + stream_iterator=response, + file_id=file_id, + model=model, + custom_llm_provider=custom_llm_provider, + logging_obj=logging_obj, + ) + except Exception as e: + raise e \ No newline at end of file diff --git a/litellm/files/streaming.py b/litellm/files/streaming.py new file mode 100644 index 0000000000..60b8df2294 --- /dev/null +++ b/litellm/files/streaming.py @@ -0,0 +1,205 @@ +import datetime +import traceback +from typing import AsyncIterator, Dict, Iterator, Literal, Optional, Union, cast + +from litellm.litellm_core_utils.litellm_logging import ( + Logging as LiteLLMLoggingObj, + get_standard_logging_object_payload, +) +from litellm.types.utils import StandardLoggingHiddenParams, StandardLoggingPayload + +FileContentProvider = Literal[ + "openai", "azure", "vertex_ai", "bedrock", "hosted_vllm", "anthropic", "manus" +] + + +class FileContentStreamingResponse: + """ + Iterator wrapper for file content streaming that carries LiteLLM metadata + and emits success/failure callbacks once the stream finishes. + """ + + def __init__( + self, + stream_iterator: Union[Iterator[bytes], AsyncIterator[bytes]], + file_id: str, + model: Optional[str], + custom_llm_provider: Optional[Union[FileContentProvider, str]], + logging_obj: Optional[LiteLLMLoggingObj], + ) -> None: + self.stream_iterator = stream_iterator + self.file_id = file_id + self.model = model + self.custom_llm_provider = custom_llm_provider + self.logging_obj = logging_obj + self.standard_logging_object: Optional[StandardLoggingPayload] = None + self._hidden_params: StandardLoggingHiddenParams = cast( + StandardLoggingHiddenParams, {} + ) + self._logging_completed = False + self._start_time = ( + logging_obj.start_time + if logging_obj is not None and getattr(logging_obj, "start_time", None) + else datetime.datetime.now() + ) + + def __iter__(self) -> "FileContentStreamingResponse": + if not hasattr(self.stream_iterator, "__next__"): + raise TypeError("File content stream does not support sync iteration") + return self + + def __next__(self) -> bytes: + if not hasattr(self.stream_iterator, "__next__"): + raise TypeError("File content stream does not support sync iteration") + + try: + return next(cast(Iterator[bytes], self.stream_iterator)) + except StopIteration: + self._log_success_sync() + raise + except Exception as e: + self._log_failure_sync(e) + raise + + def __aiter__(self) -> "FileContentStreamingResponse": + if not hasattr(self.stream_iterator, "__anext__"): + raise TypeError("File content stream does not support async iteration") + return self + + async def __anext__(self) -> bytes: + if not hasattr(self.stream_iterator, "__anext__"): + raise TypeError("File content stream does not support async iteration") + + try: + return await cast(AsyncIterator[bytes], self.stream_iterator).__anext__() + except StopAsyncIteration: + await self._log_success_async() + raise + except Exception as e: + await self._log_failure_async(e) + raise + + def _build_logging_response(self) -> Dict[str, str]: + response = { + "id": self.file_id, + "object": "file.content", + } + if self.model: + response["model"] = self.model + return response + + def _sync_hidden_params(self) -> None: + litellm_params = {} + if self.logging_obj is not None: + litellm_params = ( + self.logging_obj.model_call_details.get("litellm_params", {}) or {} + ) + + if "api_base" not in self._hidden_params and litellm_params.get("api_base"): + self._hidden_params["api_base"] = litellm_params["api_base"] + + # The generic client decorator infers `model` from the first positional arg, + # which is `file_id` for this API. Correct it before logging callbacks run. + self._hidden_params["litellm_model_name"] = self.model + if "response_cost" not in self._hidden_params: + self._hidden_params["response_cost"] = None + + def _build_standard_logging_object( + self, + end_time: datetime.datetime, + ) -> Optional[StandardLoggingPayload]: + if self.standard_logging_object is not None: + return self.standard_logging_object + + if self.logging_obj is None: + return None + + self._sync_hidden_params() + payload = get_standard_logging_object_payload( + kwargs=self.logging_obj.model_call_details, + init_response_obj=self._build_logging_response(), + start_time=self._start_time, + end_time=end_time, + logging_obj=self.logging_obj, + status="success", + ) + if payload is None: + return None + + merged_hidden_params = cast( + StandardLoggingHiddenParams, + { + **cast( + StandardLoggingHiddenParams, payload.get("hidden_params") or {} + ), + **self._hidden_params, + }, + ) + payload["hidden_params"] = merged_hidden_params + payload["response"] = self._build_logging_response() + if self.custom_llm_provider is not None: + payload["custom_llm_provider"] = self.custom_llm_provider + if self.model is not None: + payload["model"] = self.model + if self._hidden_params.get("api_base"): + payload["api_base"] = cast(str, self._hidden_params["api_base"]) + + self.standard_logging_object = payload + return payload + + async def _log_success_async(self) -> None: + if self._logging_completed or self.logging_obj is None: + return + + self._logging_completed = True + end_time = datetime.datetime.now() + standard_logging_object = self._build_standard_logging_object(end_time=end_time) + await self.logging_obj.async_success_handler( + result=self._build_logging_response(), + start_time=self._start_time, + end_time=end_time, + standard_logging_object=standard_logging_object, + ) + self.logging_obj.handle_sync_success_callbacks_for_async_calls( + result=self._build_logging_response(), + start_time=self._start_time, + end_time=end_time, + ) + + def _log_success_sync(self) -> None: + if self._logging_completed or self.logging_obj is None: + return + + self._logging_completed = True + end_time = datetime.datetime.now() + standard_logging_object = self._build_standard_logging_object(end_time=end_time) + self.logging_obj.success_handler( + result=self._build_logging_response(), + start_time=self._start_time, + end_time=end_time, + standard_logging_object=standard_logging_object, + ) + + async def _log_failure_async(self, error: Exception) -> None: + if self._logging_completed or self.logging_obj is None: + return + + self._logging_completed = True + end_time = datetime.datetime.now() + traceback_str = traceback.format_exc() + self.logging_obj.failure_handler( + error, traceback_str, self._start_time, end_time + ) + await self.logging_obj.async_failure_handler( + error, traceback_str, self._start_time, end_time + ) + + def _log_failure_sync(self, error: Exception) -> None: + if self._logging_completed or self.logging_obj is None: + return + + self._logging_completed = True + end_time = datetime.datetime.now() + self.logging_obj.failure_handler( + error, traceback.format_exc(), self._start_time, end_time + ) diff --git a/litellm/llms/openai/openai.py b/litellm/llms/openai/openai.py index be54267748..c42305ac6b 100644 --- a/litellm/llms/openai/openai.py +++ b/litellm/llms/openai/openai.py @@ -1751,6 +1751,63 @@ class OpenAIFilesAPI(BaseLLM): return HttpxBinaryResponseContent(response=response.response) + async def afile_content_streaming( + self, + file_content_request: FileContentRequest, + openai_client: AsyncOpenAI, + chunk_size: int = 1024 * 1024, + ) -> AsyncIterator[bytes]: + async with openai_client.files.with_streaming_response.content( + **file_content_request + ) as response: + async for chunk in response.iter_bytes(chunk_size=chunk_size): + yield chunk + + def file_content_streaming( + self, + _is_async: bool, + file_content_request: FileContentRequest, + api_base: str, + api_key: Optional[str], + timeout: Union[float, httpx.Timeout], + max_retries: Optional[int], + organization: Optional[str], + chunk_size: int = 1024 * 1024, + client: Optional[Union[OpenAI, AsyncOpenAI]] = None, + ) -> Union[Iterator[bytes], AsyncIterator[bytes]]: + openai_client: Optional[Union[OpenAI, AsyncOpenAI]] = self.get_openai_client( + api_key=api_key, + api_base=api_base, + timeout=timeout, + max_retries=max_retries, + organization=organization, + client=client, + _is_async=_is_async, + ) + if openai_client is None: + raise ValueError( + "OpenAI client is not initialized. Make sure api_key is passed or OPENAI_API_KEY is set in the environment." + ) + + if _is_async is True: + if not isinstance(openai_client, AsyncOpenAI): + raise ValueError( + "OpenAI client is not an instance of AsyncOpenAI. Make sure you passed an AsyncOpenAI client." + ) + return self.afile_content_streaming( # type: ignore + file_content_request=file_content_request, + openai_client=openai_client, + chunk_size=chunk_size, + ) + + def _stream() -> Iterator[bytes]: + with cast(OpenAI, openai_client).files.with_streaming_response.content( + **file_content_request + ) as response: + yield from response.iter_bytes(chunk_size=chunk_size) + + return _stream() + async def aretrieve_file( self, file_id: str, @@ -3045,4 +3102,4 @@ class OpenAIAssistantsAPI(BaseLLM): tools=tools, ) - return response + return response \ No newline at end of file diff --git a/litellm/proxy/openai_files_endpoints/files_endpoints.py b/litellm/proxy/openai_files_endpoints/files_endpoints.py index 973836b13d..05b1172158 100644 --- a/litellm/proxy/openai_files_endpoints/files_endpoints.py +++ b/litellm/proxy/openai_files_endpoints/files_endpoints.py @@ -7,7 +7,7 @@ import asyncio import traceback -from typing import Any, Optional, cast, get_args +from typing import Any, AsyncIterator, Optional, cast, get_args import httpx from fastapi import ( @@ -21,6 +21,7 @@ from fastapi import ( UploadFile, status, ) +from fastapi.responses import StreamingResponse import litellm from litellm import CreateFileRequest, get_secret_str @@ -62,6 +63,88 @@ router = APIRouter() files_config = None +def _should_stream_file_content( + *, + custom_llm_provider: str, + is_base64_unified_file_id: Any, +) -> bool: + return ( + custom_llm_provider == "openai" + and bool(is_base64_unified_file_id) is False + ) + + +async def _stream_file_content_with_logging( + stream_iterator: AsyncIterator[bytes], + proxy_logging_obj: ProxyLogging, + user_api_key_dict: UserAPIKeyAuth, + data: Dict[str, Any], +): + try: + async for chunk in stream_iterator: + yield chunk + await proxy_logging_obj.update_request_status( + litellm_call_id=data.get("litellm_call_id", ""), status="success" + ) + except Exception as e: + await proxy_logging_obj.post_call_failure_hook( + user_api_key_dict=user_api_key_dict, + original_exception=e, + request_data=data, + ) + raise + + +async def _get_streaming_file_content_response( + *, + custom_llm_provider: str, + file_id: str, + data: Dict[str, Any], + should_route: bool, + original_file_id: Optional[str], + credentials: Optional[Dict[str, Any]], + proxy_logging_obj: ProxyLogging, + user_api_key_dict: UserAPIKeyAuth, + version: str, +) -> StreamingResponse: + if should_route: + prepare_data_with_credentials( + data=data, + credentials=credentials, # type: ignore[arg-type] + file_id=original_file_id, + ) + + stream_iterator = cast( + AsyncIterator[bytes], + await litellm.afile_content_streaming( + **{ + "custom_llm_provider": custom_llm_provider, + "file_id": file_id, + **data, + } # type: ignore + ), + ) + hidden_params = getattr(stream_iterator, "_hidden_params", {}) or {} + response_headers = ProxyBaseLLMRequestProcessing.get_custom_headers( + user_api_key_dict=user_api_key_dict, + model_id=hidden_params.get("model_id", "") or "", + cache_key=hidden_params.get("cache_key", "") or "", + api_base=hidden_params.get("api_base", "") or "", + version=version, + model_region=getattr(user_api_key_dict, "allowed_model_region", ""), + ) + return StreamingResponse( + _stream_file_content_with_logging( + stream_iterator=stream_iterator, + proxy_logging_obj=proxy_logging_obj, + user_api_key_dict=user_api_key_dict, + data=data, + ), + media_type="application/octet-stream", + headers=response_headers, + ) + + def set_files_config(config): global files_config if config is None: @@ -633,7 +716,7 @@ async def get_file_content( # noqa: PLR0915 or await get_custom_llm_provider_from_request_body(request=request) or "openai" ) - + ## check if file_id is a litellm managed file is_base64_unified_file_id = _is_base64_encoded_unified_file_id(file_id) if is_base64_unified_file_id: @@ -731,6 +814,25 @@ async def get_file_content( # noqa: PLR0915 check_file_id_encoding=True, ) + if _should_stream_file_content( + custom_llm_provider=custom_llm_provider, + is_base64_unified_file_id=is_base64_unified_file_id, + ): + verbose_proxy_logger.debug( + "Routing file content request to streaming response helper" + ) + return await _get_streaming_file_content_response( + custom_llm_provider=custom_llm_provider, + file_id=file_id, + data=data, + should_route=should_route, + original_file_id=original_file_id, + credentials=credentials, + proxy_logging_obj=proxy_logging_obj, + user_api_key_dict=user_api_key_dict, + version=version, + ) + if should_route: # Use model-based routing with credentials from config prepare_data_with_credentials( @@ -738,7 +840,7 @@ async def get_file_content( # noqa: PLR0915 credentials=credentials, # type: ignore file_id=original_file_id, # Use decoded file ID if from encoded ID ) - + response = await litellm.afile_content( custom_llm_provider=credentials["custom_llm_provider"], # type: ignore **data, diff --git a/litellm/utils.py b/litellm/utils.py index f902644e76..f0f0e231c1 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -2133,6 +2133,8 @@ def _is_async_request( _STREAMING_CALL_TYPES = frozenset( { + "afile_content_streaming", + "file_content_streaming", CallTypes.generate_content_stream, CallTypes.agenerate_content_stream, CallTypes.generate_content_stream.value, diff --git a/tests/test_litellm/llms/openai/test_openai_file_content_streaming.py b/tests/test_litellm/llms/openai/test_openai_file_content_streaming.py new file mode 100644 index 0000000000..3362af82a3 --- /dev/null +++ b/tests/test_litellm/llms/openai/test_openai_file_content_streaming.py @@ -0,0 +1,102 @@ +import pytest +from typing import AsyncIterator, cast + +from litellm.files import main as files_main +from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj + + +@pytest.mark.asyncio +async def test_afile_content_streaming_routes_to_openai_streaming_handler( + monkeypatch, +): + captured_kwargs = {} + + async def _mock_stream(): + yield b"hello " + yield b"world" + + def _mock_file_content_streaming(**kwargs): + captured_kwargs.update(kwargs) + return _mock_stream() + + monkeypatch.setattr( + files_main.openai_files_instance, + "file_content_streaming", + _mock_file_content_streaming, + ) + + stream_iterator = await files_main.afile_content_streaming( + file_id="file-abc123", + custom_llm_provider="openai", + api_key="sk-test", + api_base="https://api.openai.com/v1", + organization="org-123", + chunk_size=8, + ) + + async_stream_iterator = cast(AsyncIterator[bytes], stream_iterator) + chunks = [chunk async for chunk in async_stream_iterator] + + assert chunks == [b"hello ", b"world"] + assert captured_kwargs["_is_async"] is True + assert captured_kwargs["file_content_request"]["file_id"] == "file-abc123" + assert captured_kwargs["api_key"] == "sk-test" + assert captured_kwargs["api_base"] == "https://api.openai.com/v1" + assert captured_kwargs["organization"] == "org-123" + assert captured_kwargs["chunk_size"] == 8 + + +@pytest.mark.asyncio +async def test_afile_content_streaming_builds_standard_logging_object_on_completion( + monkeypatch, +): + captured_standard_logging_object = None + + async def _mock_stream(): + yield b"hello" + + def _mock_file_content_streaming(**kwargs): + return _mock_stream() + + async def _mock_async_success_handler( + self, result=None, start_time=None, end_time=None, cache_hit=None, **kwargs + ): + nonlocal captured_standard_logging_object + captured_standard_logging_object = kwargs.get("standard_logging_object") + self.model_call_details["standard_logging_object"] = captured_standard_logging_object + + monkeypatch.setattr( + files_main.openai_files_instance, + "file_content_streaming", + _mock_file_content_streaming, + ) + monkeypatch.setattr( + LiteLLMLoggingObj, + "async_success_handler", + _mock_async_success_handler, + ) + monkeypatch.setattr( + LiteLLMLoggingObj, + "handle_sync_success_callbacks_for_async_calls", + lambda self, result, start_time, end_time, cache_hit=None: None, + ) + + stream_iterator = await files_main.afile_content_streaming( + file_id="file-abc123", + custom_llm_provider="openai", + api_key="sk-test", + api_base="https://api.openai.com/v1", + ) + + async_stream_iterator = cast(AsyncIterator[bytes], stream_iterator) + chunks = [chunk async for chunk in async_stream_iterator] + + assert chunks == [b"hello"] + assert captured_standard_logging_object is not None + assert captured_standard_logging_object["call_type"] == "afile_content_streaming" + assert captured_standard_logging_object["custom_llm_provider"] == "openai" + assert captured_standard_logging_object["response"]["id"] == "file-abc123" + assert ( + captured_standard_logging_object["hidden_params"]["api_base"] + == "https://api.openai.com/v1" + ) diff --git a/tests/test_litellm/proxy/conftest.py b/tests/test_litellm/proxy/conftest.py index d7cf82d641..d30859706e 100644 --- a/tests/test_litellm/proxy/conftest.py +++ b/tests/test_litellm/proxy/conftest.py @@ -13,6 +13,30 @@ import pytest import yaml from fastapi.testclient import TestClient +def _patch_missing_responses_activate() -> None: + """ + Ensure tests using @responses.activate are skipped when the installed + `responses` package does not expose `activate`. + """ + try: + import responses # type: ignore + except Exception: + return + + if hasattr(responses, "activate"): + return + + reason = "Skipping: installed responses package has no 'activate' attribute" + + def _skip_activate(func=None, *args, **kwargs): + if func is None: + return lambda f: pytest.mark.skip(reason=reason)(f) + return pytest.mark.skip(reason=reason)(func) + + setattr(responses, "activate", _skip_activate) + + +_patch_missing_responses_activate() def build_cache_config(enable_cache: bool = True) -> Optional[Dict]: """ diff --git a/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py b/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py index c6a03cf4ec..9e27992024 100644 --- a/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py +++ b/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py @@ -1552,3 +1552,59 @@ def test_file_invalid_anchor_returns_500( ) assert response.status_code == 500 assert "created_at" in response.json()["error"]["message"] + + +def test_get_file_content_streams_openai_direct_path( + mocker: MockerFixture, monkeypatch, llm_router: Router +): + import litellm.proxy.proxy_server as ps + from litellm.proxy._types import LitellmUserRoles + + proxy_logging_obj = setup_proxy_logging_object(monkeypatch, llm_router) + monkeypatch.setattr("litellm.proxy.proxy_server.master_key", None) + monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", None) + proxy_logging_obj.update_request_status = mocker.AsyncMock() + proxy_logging_obj.post_call_failure_hook = mocker.AsyncMock() + + captured_kwargs = {} + + async def _mock_afile_content_streaming(**kwargs): + captured_kwargs.update(kwargs) + + async def _stream(): + yield b"hello " + yield b"world" + + return _stream() + + async def _fail_buffered_path(*args, **kwargs): + raise AssertionError("buffered afile_content path should not be used") + + monkeypatch.setattr(litellm, "afile_content_streaming", _mock_afile_content_streaming) + monkeypatch.setattr(litellm, "afile_content", _fail_buffered_path) + monkeypatch.setattr( + "litellm.proxy.openai_files_endpoints.files_endpoints.handle_model_based_routing", + lambda **kwargs: (False, None, None, None), + ) + + app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth( + api_key="test-key", + user_role=LitellmUserRoles.PROXY_ADMIN, + user_id="test-user", + ) + + try: + response = client.get( + "/v1/files/file-abc123/content", + headers={"Authorization": "Bearer test-key"}, + ) + finally: + app.dependency_overrides.pop(ps.user_api_key_auth, None) + + assert response.status_code == 200, response.text + assert response.content == b"hello world" + assert response.headers["content-type"].startswith("application/octet-stream") + assert captured_kwargs["custom_llm_provider"] == "openai" + assert captured_kwargs["file_id"] == "file-abc123" + proxy_logging_obj.update_request_status.assert_awaited_once() + proxy_logging_obj.post_call_failure_hook.assert_not_called() From 13108039c85389d2a635c0be3682056c452fd759 Mon Sep 17 00:00:00 2001 From: harish876 Date: Thu, 9 Apr 2026 22:29:49 +0000 Subject: [PATCH 02/14] remove conftest patch. TODO: make a different PR for this --- tests/test_litellm/proxy/conftest.py | 25 ------------------------- 1 file changed, 25 deletions(-) diff --git a/tests/test_litellm/proxy/conftest.py b/tests/test_litellm/proxy/conftest.py index d30859706e..a5d8fd1707 100644 --- a/tests/test_litellm/proxy/conftest.py +++ b/tests/test_litellm/proxy/conftest.py @@ -13,31 +13,6 @@ import pytest import yaml from fastapi.testclient import TestClient -def _patch_missing_responses_activate() -> None: - """ - Ensure tests using @responses.activate are skipped when the installed - `responses` package does not expose `activate`. - """ - try: - import responses # type: ignore - except Exception: - return - - if hasattr(responses, "activate"): - return - - reason = "Skipping: installed responses package has no 'activate' attribute" - - def _skip_activate(func=None, *args, **kwargs): - if func is None: - return lambda f: pytest.mark.skip(reason=reason)(f) - return pytest.mark.skip(reason=reason)(func) - - setattr(responses, "activate", _skip_activate) - - -_patch_missing_responses_activate() - def build_cache_config(enable_cache: bool = True) -> Optional[Dict]: """ Build Redis cache configuration from environment variables. From af4d4ab2ee4f410cb1d54d9dc5b77b21bad27d72 Mon Sep 17 00:00:00 2001 From: harish876 Date: Fri, 10 Apr 2026 06:54:08 +0000 Subject: [PATCH 03/14] Introduced Content-Length response headers into the streaming response. This provides a 1:1 behaviour mapping similar to the non streaming behaviour. --- litellm/files/main.py | 41 ++++++++++++++----- litellm/files/streaming.py | 7 +++- litellm/llms/openai/openai.py | 40 ++++++++++++------ .../openai_files_endpoints/files_endpoints.py | 35 +++++++++------- .../test_openai_file_content_streaming.py | 21 +++++++--- .../test_files_endpoint.py | 7 +++- 6 files changed, 105 insertions(+), 46 deletions(-) diff --git a/litellm/files/main.py b/litellm/files/main.py index 865b679e04..2c06f9459e 100644 --- a/litellm/files/main.py +++ b/litellm/files/main.py @@ -36,7 +36,10 @@ FileContentProvider = Literal[ import litellm from litellm import get_secret_str -from litellm.files.streaming import FileContentStreamingResponse +from litellm.files.streaming import ( + FileContentStreamingResponse, + FileContentStreamingResult, +) from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.llms.azure.common_utils import get_azure_credentials @@ -990,7 +993,7 @@ async def afile_content_streaming( extra_body: Optional[Dict[str, str]] = None, chunk_size: int = 1024 * 1024, **kwargs, -) -> Union[Iterator[bytes], AsyncIterator[bytes]]: +) -> FileContentStreamingResult: """ Async wrapper for file_content_streaming. """ @@ -1034,7 +1037,7 @@ def file_content_streaming( extra_body: Optional[Dict[str, str]] = None, chunk_size: int = 1024 * 1024, **kwargs, -) -> Union[Iterator[bytes], AsyncIterator[bytes]]: +) -> Union[FileContentStreamingResult, Coroutine[Any, Any, FileContentStreamingResult]]: """ Prototype API: Returns a byte iterator for file contents. @@ -1080,7 +1083,23 @@ def file_content_streaming( litellm_params["api_base"] = optional_params.api_base logging_obj.model_call_details["litellm_params"] = litellm_params - response = cast(Union[Iterator[bytes], AsyncIterator[bytes]], iter(())) + def _wrap_streaming_result( + response: FileContentStreamingResult, + ) -> FileContentStreamingResult: + return FileContentStreamingResult( + stream_iterator=FileContentStreamingResponse( + stream_iterator=response.stream_iterator, + file_id=file_id, + model=model, + custom_llm_provider=custom_llm_provider, + logging_obj=logging_obj, + ), + headers=response.headers, + ) + + response: Union[ + FileContentStreamingResult, Coroutine[Any, Any, FileContentStreamingResult] + ] = FileContentStreamingResult(stream_iterator=iter(()), headers={}) if custom_llm_provider in OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS: openai_creds = get_openai_credentials( api_base=optional_params.api_base, @@ -1115,12 +1134,12 @@ def file_content_streaming( ), ) - return FileContentStreamingResponse( - stream_iterator=response, - file_id=file_id, - model=model, - custom_llm_provider=custom_llm_provider, - logging_obj=logging_obj, - ) + if asyncio.iscoroutine(response): + async def _await_and_wrap() -> FileContentStreamingResult: + return _wrap_streaming_result(await response) + + return _await_and_wrap() + + return _wrap_streaming_result(response) except Exception as e: raise e \ No newline at end of file diff --git a/litellm/files/streaming.py b/litellm/files/streaming.py index 60b8df2294..ab149cdb4c 100644 --- a/litellm/files/streaming.py +++ b/litellm/files/streaming.py @@ -1,6 +1,6 @@ import datetime import traceback -from typing import AsyncIterator, Dict, Iterator, Literal, Optional, Union, cast +from typing import AsyncIterator, Dict, Iterator, Literal, NamedTuple, Optional, Union, cast from litellm.litellm_core_utils.litellm_logging import ( Logging as LiteLLMLoggingObj, @@ -13,6 +13,11 @@ FileContentProvider = Literal[ ] +class FileContentStreamingResult(NamedTuple): + stream_iterator: Union[Iterator[bytes], AsyncIterator[bytes]] + headers: Dict[str, str] + + class FileContentStreamingResponse: """ Iterator wrapper for file content streaming that carries LiteLLM metadata diff --git a/litellm/llms/openai/openai.py b/litellm/llms/openai/openai.py index c42305ac6b..086ee84852 100644 --- a/litellm/llms/openai/openai.py +++ b/litellm/llms/openai/openai.py @@ -32,6 +32,7 @@ import litellm from litellm import LlmProviders from litellm._logging import verbose_logger from litellm.constants import DEFAULT_MAX_RETRIES +from litellm.files.streaming import FileContentStreamingResult from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.litellm_core_utils.logging_utils import track_llm_api_timing from litellm.llms.base_llm.base_model_iterator import BaseModelResponseIterator @@ -1756,12 +1757,21 @@ class OpenAIFilesAPI(BaseLLM): file_content_request: FileContentRequest, openai_client: AsyncOpenAI, chunk_size: int = 1024 * 1024, - ) -> AsyncIterator[bytes]: - async with openai_client.files.with_streaming_response.content( + ) -> FileContentStreamingResult: + response_cm = openai_client.files.with_streaming_response.content( **file_content_request - ) as response: - async for chunk in response.iter_bytes(chunk_size=chunk_size): - yield chunk + ) + response = await response_cm.__aenter__() + headers = dict(response.headers) + + async def _stream() -> AsyncIterator[bytes]: + try: + async for chunk in response.iter_bytes(chunk_size=chunk_size): + yield chunk + finally: + await response_cm.__aexit__(None, None, None) + + return FileContentStreamingResult(stream_iterator=_stream(), headers=headers) def file_content_streaming( self, @@ -1774,7 +1784,7 @@ class OpenAIFilesAPI(BaseLLM): organization: Optional[str], chunk_size: int = 1024 * 1024, client: Optional[Union[OpenAI, AsyncOpenAI]] = None, - ) -> Union[Iterator[bytes], AsyncIterator[bytes]]: + ) -> FileContentStreamingResult: openai_client: Optional[Union[OpenAI, AsyncOpenAI]] = self.get_openai_client( api_key=api_key, api_base=api_base, @@ -1800,13 +1810,19 @@ class OpenAIFilesAPI(BaseLLM): chunk_size=chunk_size, ) - def _stream() -> Iterator[bytes]: - with cast(OpenAI, openai_client).files.with_streaming_response.content( - **file_content_request - ) as response: - yield from response.iter_bytes(chunk_size=chunk_size) + response_cm = cast(OpenAI, openai_client).files.with_streaming_response.content( + **file_content_request + ) + response = response_cm.__enter__() + headers = dict(response.headers) - return _stream() + def _stream() -> Iterator[bytes]: + try: + yield from response.iter_bytes(chunk_size=chunk_size) + finally: + response_cm.__exit__(None, None, None) + + return FileContentStreamingResult(stream_iterator=_stream(), headers=headers) async def aretrieve_file( self, diff --git a/litellm/proxy/openai_files_endpoints/files_endpoints.py b/litellm/proxy/openai_files_endpoints/files_endpoints.py index 05b1172158..acb2ffc3fe 100644 --- a/litellm/proxy/openai_files_endpoints/files_endpoints.py +++ b/litellm/proxy/openai_files_endpoints/files_endpoints.py @@ -114,25 +114,30 @@ async def _get_streaming_file_content_response( file_id=original_file_id, ) + stream_result = await litellm.afile_content_streaming( + **{ + "custom_llm_provider": custom_llm_provider, + "file_id": file_id, + **data, + } # type: ignore + ) stream_iterator = cast( AsyncIterator[bytes], - await litellm.afile_content_streaming( - **{ - "custom_llm_provider": custom_llm_provider, - "file_id": file_id, - **data, - } # type: ignore - ), + stream_result.stream_iterator, ) hidden_params = getattr(stream_iterator, "_hidden_params", {}) or {} - response_headers = ProxyBaseLLMRequestProcessing.get_custom_headers( - user_api_key_dict=user_api_key_dict, - model_id=hidden_params.get("model_id", "") or "", - cache_key=hidden_params.get("cache_key", "") or "", - api_base=hidden_params.get("api_base", "") or "", - version=version, - model_region=getattr(user_api_key_dict, "allowed_model_region", ""), - ) + response_headers = { + **stream_result.headers, + **ProxyBaseLLMRequestProcessing.get_custom_headers( + user_api_key_dict=user_api_key_dict, + model_id=hidden_params.get("model_id", "") or "", + cache_key=hidden_params.get("cache_key", "") or "", + api_base=hidden_params.get("api_base", "") or "", + version=version, + model_region=getattr(user_api_key_dict, "allowed_model_region", ""), + ), + } + return StreamingResponse( _stream_file_content_with_logging( stream_iterator=stream_iterator, diff --git a/tests/test_litellm/llms/openai/test_openai_file_content_streaming.py b/tests/test_litellm/llms/openai/test_openai_file_content_streaming.py index 3362af82a3..fcc0d9a97b 100644 --- a/tests/test_litellm/llms/openai/test_openai_file_content_streaming.py +++ b/tests/test_litellm/llms/openai/test_openai_file_content_streaming.py @@ -2,6 +2,7 @@ import pytest from typing import AsyncIterator, cast from litellm.files import main as files_main +from litellm.files.streaming import FileContentStreamingResult from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj @@ -17,7 +18,10 @@ async def test_afile_content_streaming_routes_to_openai_streaming_handler( def _mock_file_content_streaming(**kwargs): captured_kwargs.update(kwargs) - return _mock_stream() + return FileContentStreamingResult( + stream_iterator=_mock_stream(), + headers={"content-length": "11"}, + ) monkeypatch.setattr( files_main.openai_files_instance, @@ -25,7 +29,7 @@ async def test_afile_content_streaming_routes_to_openai_streaming_handler( _mock_file_content_streaming, ) - stream_iterator = await files_main.afile_content_streaming( + stream_result = await files_main.afile_content_streaming( file_id="file-abc123", custom_llm_provider="openai", api_key="sk-test", @@ -34,10 +38,11 @@ async def test_afile_content_streaming_routes_to_openai_streaming_handler( chunk_size=8, ) - async_stream_iterator = cast(AsyncIterator[bytes], stream_iterator) + async_stream_iterator = cast(AsyncIterator[bytes], stream_result.stream_iterator) chunks = [chunk async for chunk in async_stream_iterator] assert chunks == [b"hello ", b"world"] + assert stream_result.headers["content-length"] == "11" assert captured_kwargs["_is_async"] is True assert captured_kwargs["file_content_request"]["file_id"] == "file-abc123" assert captured_kwargs["api_key"] == "sk-test" @@ -56,7 +61,10 @@ async def test_afile_content_streaming_builds_standard_logging_object_on_complet yield b"hello" def _mock_file_content_streaming(**kwargs): - return _mock_stream() + return FileContentStreamingResult( + stream_iterator=_mock_stream(), + headers={"content-length": "5"}, + ) async def _mock_async_success_handler( self, result=None, start_time=None, end_time=None, cache_hit=None, **kwargs @@ -81,17 +89,18 @@ async def test_afile_content_streaming_builds_standard_logging_object_on_complet lambda self, result, start_time, end_time, cache_hit=None: None, ) - stream_iterator = await files_main.afile_content_streaming( + stream_result = await files_main.afile_content_streaming( file_id="file-abc123", custom_llm_provider="openai", api_key="sk-test", api_base="https://api.openai.com/v1", ) - async_stream_iterator = cast(AsyncIterator[bytes], stream_iterator) + async_stream_iterator = cast(AsyncIterator[bytes], stream_result.stream_iterator) chunks = [chunk async for chunk in async_stream_iterator] assert chunks == [b"hello"] + assert stream_result.headers["content-length"] == "5" assert captured_standard_logging_object is not None assert captured_standard_logging_object["call_type"] == "afile_content_streaming" assert captured_standard_logging_object["custom_llm_provider"] == "openai" diff --git a/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py b/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py index 9e27992024..31a6b37437 100644 --- a/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py +++ b/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py @@ -14,6 +14,7 @@ sys.path.insert( import litellm from litellm import Router +from litellm.files.streaming import FileContentStreamingResult from litellm.proxy._types import LiteLLM_UserTableFiltered, UserAPIKeyAuth from litellm.proxy.hooks import get_proxy_hook from litellm.proxy.management_endpoints.internal_user_endpoints import ui_view_users @@ -1575,7 +1576,10 @@ def test_get_file_content_streams_openai_direct_path( yield b"hello " yield b"world" - return _stream() + return FileContentStreamingResult( + stream_iterator=_stream(), + headers={"content-length": "11"}, + ) async def _fail_buffered_path(*args, **kwargs): raise AssertionError("buffered afile_content path should not be used") @@ -1604,6 +1608,7 @@ def test_get_file_content_streams_openai_direct_path( assert response.status_code == 200, response.text assert response.content == b"hello world" assert response.headers["content-type"].startswith("application/octet-stream") + assert response.headers["content-length"] == "11" assert captured_kwargs["custom_llm_provider"] == "openai" assert captured_kwargs["file_id"] == "file-abc123" proxy_logging_obj.update_request_status.assert_awaited_once() From 044d434b502de6bcfd853a21c52efbb034b8f315 Mon Sep 17 00:00:00 2001 From: harish876 Date: Fri, 10 Apr 2026 07:11:31 +0000 Subject: [PATCH 04/14] remove unused iterator imports --- litellm/files/main.py | 12 +----------- 1 file changed, 1 insertion(+), 11 deletions(-) diff --git a/litellm/files/main.py b/litellm/files/main.py index 2c06f9459e..8716cf3b0f 100644 --- a/litellm/files/main.py +++ b/litellm/files/main.py @@ -10,7 +10,7 @@ import contextvars import time import uuid as uuid_module from functools import partial -from typing import Any, AsyncIterator, Coroutine, Dict, Iterator, Literal, Optional, Union, cast +from typing import Any,Coroutine, Dict, Literal, Optional, Union, cast import httpx @@ -1045,20 +1045,10 @@ def file_content_streaming( """ try: optional_params = GenericLiteLLMParams(**kwargs) - litellm_params_dict = get_litellm_params(**kwargs) - client = kwargs.get("client") logging_obj = cast( Optional[LiteLLMLoggingObj], kwargs.get("litellm_logging_obj") ) - try: - if model is not None: - _, custom_llm_provider, _, _ = get_llm_provider( - model, custom_llm_provider - ) - except Exception: - pass - timeout = optional_params.timeout or kwargs.get("request_timeout", 600) or 600 if ( timeout is not None From baba3ebed896acbf73d6fc50c86f7ae0f9f27d90 Mon Sep 17 00:00:00 2001 From: harish876 Date: Fri, 10 Apr 2026 18:30:28 +0000 Subject: [PATCH 05/14] Refactor file content streaming implementation - Removed unused imports and streamlined type hints in `litellm/utils.py` and `litellm/files/main.py`. - Moved `FileContentStreamingResult` to a new `litellm/files/types.py` for better organization. - Updated `FileContentStreamingResponse` in `litellm/files/streaming.py` to include asynchronous close methods and improved logging capabilities. - Enhanced tests to ensure proper closure of streaming iterators in `tests/test_litellm/llms/openai/test_openai_file_content_streaming.py` and `tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py`. --- litellm/files/main.py | 6 +- litellm/files/streaming.py | 71 +++++++++++++------ litellm/files/types.py | 6 ++ litellm/llms/openai/openai.py | 2 +- .../openai_files_endpoints/files_endpoints.py | 3 + litellm/utils.py | 2 - .../test_openai_file_content_streaming.py | 30 +++++++- .../test_files_endpoint.py | 41 ++++++++++- 8 files changed, 130 insertions(+), 31 deletions(-) create mode 100644 litellm/files/types.py diff --git a/litellm/files/main.py b/litellm/files/main.py index 8716cf3b0f..13abdfd92f 100644 --- a/litellm/files/main.py +++ b/litellm/files/main.py @@ -36,10 +36,8 @@ FileContentProvider = Literal[ import litellm from litellm import get_secret_str -from litellm.files.streaming import ( - FileContentStreamingResponse, - FileContentStreamingResult, -) +from litellm.files.streaming import FileContentStreamingResponse +from litellm.files.types import FileContentStreamingResult from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.llms.azure.common_utils import get_azure_credentials diff --git a/litellm/files/streaming.py b/litellm/files/streaming.py index ab149cdb4c..dab190b9b3 100644 --- a/litellm/files/streaming.py +++ b/litellm/files/streaming.py @@ -1,23 +1,20 @@ import datetime import traceback -from typing import AsyncIterator, Dict, Iterator, Literal, NamedTuple, Optional, Union, cast +from typing import TYPE_CHECKING, Any, AsyncIterator, Dict, Iterator, Literal, Optional, Union, cast -from litellm.litellm_core_utils.litellm_logging import ( - Logging as LiteLLMLoggingObj, - get_standard_logging_object_payload, -) -from litellm.types.utils import StandardLoggingHiddenParams, StandardLoggingPayload +import anyio + +if TYPE_CHECKING: + from litellm.litellm_core_utils.litellm_logging import ( + Logging as LiteLLMLoggingObj, + ) + from litellm.types.utils import StandardLoggingHiddenParams, StandardLoggingPayload FileContentProvider = Literal[ "openai", "azure", "vertex_ai", "bedrock", "hosted_vllm", "anthropic", "manus" ] -class FileContentStreamingResult(NamedTuple): - stream_iterator: Union[Iterator[bytes], AsyncIterator[bytes]] - headers: Dict[str, str] - - class FileContentStreamingResponse: """ Iterator wrapper for file content streaming that carries LiteLLM metadata @@ -30,18 +27,17 @@ class FileContentStreamingResponse: file_id: str, model: Optional[str], custom_llm_provider: Optional[Union[FileContentProvider, str]], - logging_obj: Optional[LiteLLMLoggingObj], + logging_obj: Optional["LiteLLMLoggingObj"], ) -> None: self.stream_iterator = stream_iterator self.file_id = file_id self.model = model self.custom_llm_provider = custom_llm_provider self.logging_obj = logging_obj - self.standard_logging_object: Optional[StandardLoggingPayload] = None - self._hidden_params: StandardLoggingHiddenParams = cast( - StandardLoggingHiddenParams, {} - ) + self.standard_logging_object: Optional["StandardLoggingPayload"] = None + self._hidden_params: Dict[str, Any] = {} self._logging_completed = False + self._close_completed = False self._start_time = ( logging_obj.start_time if logging_obj is not None and getattr(logging_obj, "start_time", None) @@ -84,6 +80,37 @@ class FileContentStreamingResponse: await self._log_failure_async(e) raise + async def aclose(self) -> None: + if self._close_completed: + return + + self._close_completed = True + self._logging_completed = True + stream_to_close = self.stream_iterator + self.stream_iterator = cast(Union[Iterator[bytes], AsyncIterator[bytes]], iter(())) + + # Shield cleanup from request cancellation so upstream HTTP connections + # are released promptly on client disconnects. + with anyio.CancelScope(shield=True): + if hasattr(stream_to_close, "aclose"): + await cast(AsyncIterator[bytes], stream_to_close).aclose() # type: ignore[attr-defined] + elif hasattr(stream_to_close, "close"): + result = cast(Iterator[bytes], stream_to_close).close() # type: ignore[attr-defined] + if result is not None: + await result + + def close(self) -> None: + if self._close_completed: + return + + self._close_completed = True + self._logging_completed = True + stream_to_close = self.stream_iterator + self.stream_iterator = cast(Union[Iterator[bytes], AsyncIterator[bytes]], iter(())) + + if hasattr(stream_to_close, "close"): + cast(Iterator[bytes], stream_to_close).close() # type: ignore[attr-defined] + def _build_logging_response(self) -> Dict[str, str]: response = { "id": self.file_id, @@ -112,13 +139,17 @@ class FileContentStreamingResponse: def _build_standard_logging_object( self, end_time: datetime.datetime, - ) -> Optional[StandardLoggingPayload]: + ) -> Optional["StandardLoggingPayload"]: if self.standard_logging_object is not None: return self.standard_logging_object if self.logging_obj is None: return None + from litellm.litellm_core_utils.litellm_logging import ( + get_standard_logging_object_payload, + ) + self._sync_hidden_params() payload = get_standard_logging_object_payload( kwargs=self.logging_obj.model_call_details, @@ -132,11 +163,9 @@ class FileContentStreamingResponse: return None merged_hidden_params = cast( - StandardLoggingHiddenParams, + "StandardLoggingHiddenParams", { - **cast( - StandardLoggingHiddenParams, payload.get("hidden_params") or {} - ), + **cast(Dict[str, Any], payload.get("hidden_params") or {}), **self._hidden_params, }, ) diff --git a/litellm/files/types.py b/litellm/files/types.py new file mode 100644 index 0000000000..2357bd997f --- /dev/null +++ b/litellm/files/types.py @@ -0,0 +1,6 @@ +from typing import AsyncIterator, Dict, Iterator, NamedTuple, Union + + +class FileContentStreamingResult(NamedTuple): + stream_iterator: Union[Iterator[bytes], AsyncIterator[bytes]] + headers: Dict[str, str] diff --git a/litellm/llms/openai/openai.py b/litellm/llms/openai/openai.py index 086ee84852..76f300ec9c 100644 --- a/litellm/llms/openai/openai.py +++ b/litellm/llms/openai/openai.py @@ -32,7 +32,7 @@ import litellm from litellm import LlmProviders from litellm._logging import verbose_logger from litellm.constants import DEFAULT_MAX_RETRIES -from litellm.files.streaming import FileContentStreamingResult +from litellm.files.types import FileContentStreamingResult from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.litellm_core_utils.logging_utils import track_llm_api_timing from litellm.llms.base_llm.base_model_iterator import BaseModelResponseIterator diff --git a/litellm/proxy/openai_files_endpoints/files_endpoints.py b/litellm/proxy/openai_files_endpoints/files_endpoints.py index acb2ffc3fe..468e0baaa4 100644 --- a/litellm/proxy/openai_files_endpoints/files_endpoints.py +++ b/litellm/proxy/openai_files_endpoints/files_endpoints.py @@ -93,6 +93,9 @@ async def _stream_file_content_with_logging( request_data=data, ) raise + finally: + if hasattr(stream_iterator, "aclose"): + await stream_iterator.aclose() # type: ignore[attr-defined] async def _get_streaming_file_content_response( diff --git a/litellm/utils.py b/litellm/utils.py index f0f0e231c1..d8e14de637 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -243,8 +243,6 @@ from typing import ( get_args, ) -from openai import OpenAIError as OriginalError - # These are lazy loaded via __getattr__ from litellm.llms.base_llm.base_utils import ( BaseLLMModelInfo, diff --git a/tests/test_litellm/llms/openai/test_openai_file_content_streaming.py b/tests/test_litellm/llms/openai/test_openai_file_content_streaming.py index fcc0d9a97b..66c6964021 100644 --- a/tests/test_litellm/llms/openai/test_openai_file_content_streaming.py +++ b/tests/test_litellm/llms/openai/test_openai_file_content_streaming.py @@ -2,7 +2,8 @@ import pytest from typing import AsyncIterator, cast from litellm.files import main as files_main -from litellm.files.streaming import FileContentStreamingResult +from litellm.files.streaming import FileContentStreamingResponse +from litellm.files.types import FileContentStreamingResult from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj @@ -109,3 +110,30 @@ async def test_afile_content_streaming_builds_standard_logging_object_on_complet captured_standard_logging_object["hidden_params"]["api_base"] == "https://api.openai.com/v1" ) + + +@pytest.mark.asyncio +async def test_file_content_streaming_response_aclose_closes_underlying_async_generator(): + close_called = False + + async def _mock_stream(): + nonlocal close_called + try: + yield b"hello" + yield b"world" + finally: + close_called = True + + stream = FileContentStreamingResponse( + stream_iterator=_mock_stream(), + file_id="file-abc123", + model="gpt-4o", + custom_llm_provider="openai", + logging_obj=None, + ) + + assert await stream.__anext__() == b"hello" + + await stream.aclose() + + assert close_called is True diff --git a/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py b/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py index 31a6b37437..3736380124 100644 --- a/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py +++ b/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py @@ -1,7 +1,7 @@ import json import os import sys -from unittest.mock import ANY +from unittest.mock import ANY, AsyncMock import pytest import respx @@ -14,7 +14,10 @@ sys.path.insert( import litellm from litellm import Router -from litellm.files.streaming import FileContentStreamingResult +from litellm.files.types import FileContentStreamingResult +from litellm.proxy.openai_files_endpoints.files_endpoints import ( + _stream_file_content_with_logging, +) from litellm.proxy._types import LiteLLM_UserTableFiltered, UserAPIKeyAuth from litellm.proxy.hooks import get_proxy_hook from litellm.proxy.management_endpoints.internal_user_endpoints import ui_view_users @@ -78,6 +81,40 @@ def setup_proxy_logging_object(monkeypatch, llm_router: Router) -> ProxyLogging: return proxy_logging_object +@pytest.mark.asyncio +async def test_stream_file_content_with_logging_closes_inner_iterator_on_early_exit(): + class MockStreamIterator: + def __init__(self) -> None: + self._chunks = iter([b"hello", b"world"]) + self.aclose = AsyncMock() + + def __aiter__(self): + return self + + async def __anext__(self): + try: + return next(self._chunks) + except StopIteration: + raise StopAsyncIteration + + stream_iterator = MockStreamIterator() + proxy_logging_obj = AsyncMock() + + generator = _stream_file_content_with_logging( + stream_iterator=stream_iterator, + proxy_logging_obj=proxy_logging_obj, + user_api_key_dict=AsyncMock(), + data={"litellm_call_id": "call-123"}, + ) + + assert await generator.__anext__() == b"hello" + + await generator.aclose() + + stream_iterator.aclose.assert_awaited_once() + proxy_logging_obj.update_request_status.assert_not_called() + + def test_invalid_purpose(mocker: MockerFixture, monkeypatch, llm_router: Router): """ Asserts 'create_file' is called with the correct arguments From 1c74e17bed3ee52015f0308f4bc2fa472aa819d0 Mon Sep 17 00:00:00 2001 From: harish876 Date: Fri, 10 Apr 2026 18:45:00 +0000 Subject: [PATCH 06/14] E2E test to assert response headers from the openai files change --- .../openai_endpoints_tests/test_openai_files_endpoints.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/tests/openai_endpoints_tests/test_openai_files_endpoints.py b/tests/openai_endpoints_tests/test_openai_files_endpoints.py index 5299cfc537..6be692b278 100644 --- a/tests/openai_endpoints_tests/test_openai_files_endpoints.py +++ b/tests/openai_endpoints_tests/test_openai_files_endpoints.py @@ -27,8 +27,16 @@ async def test_file_operations(): get_file_content = await openai_client.files.content(file_id=uploaded_file.id) print("get_file_content=", get_file_content.content) + response = get_file_content.response assert get_file_content.content == file_content + assert response.status_code == 200 + assert response.headers.get("content-type") == "application/octet-stream" + assert response.headers.get("content-length") is not None + assert int(response.headers["content-length"]) == len(get_file_content.content) + assert response.headers.get("content-disposition") is not None + assert uploaded_file.filename in response.headers["content-disposition"] + assert response.headers.get("x-request-id") is not None # try get_file_content.write_to_file get_file_content.write_to_file("get_file_content.jsonl") From ccf3dc316133677ece53a7ad0c808508a3bbd48e Mon Sep 17 00:00:00 2001 From: harish876 Date: Fri, 10 Apr 2026 22:41:13 +0000 Subject: [PATCH 07/14] Code Comments incorporated. - Static Methods for Streaming Handler Function - Remove the afile_content_streaming wrapper function. Enabled with a stream boolean in afile_content - Cleaned up test cases after refactor --- .../files/file_content_streaming_handler.py | 109 ++++++++ litellm/files/main.py | 255 ++++++++---------- litellm/files/streaming.py | 1 + litellm/llms/openai/openai.py | 18 +- .../openai_files_endpoints/files_endpoints.py | 102 +------ litellm/utils.py | 2 - .../test_openai_file_content_streaming.py | 194 ++++++++++++- .../test_files_endpoint.py | 15 +- 8 files changed, 435 insertions(+), 261 deletions(-) create mode 100644 litellm/files/file_content_streaming_handler.py diff --git a/litellm/files/file_content_streaming_handler.py b/litellm/files/file_content_streaming_handler.py new file mode 100644 index 0000000000..dd39e1b031 --- /dev/null +++ b/litellm/files/file_content_streaming_handler.py @@ -0,0 +1,109 @@ +from typing import Any, AsyncIterator, Dict, Optional, cast + +from fastapi.responses import StreamingResponse + +import litellm +from litellm.files.types import FileContentStreamingResult +from litellm.proxy._types import UserAPIKeyAuth +from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing +from litellm.proxy.openai_files_endpoints.common_utils import ( + prepare_data_with_credentials, +) +from litellm.proxy.utils import ProxyLogging + + +class FileContentStreamingHandler: + @staticmethod + def should_stream_file_content( + *, + custom_llm_provider: str, + is_base64_unified_file_id: Any, + ) -> bool: + return ( + custom_llm_provider == "openai" + and bool(is_base64_unified_file_id) is False + ) + + @staticmethod + async def stream_file_content_with_logging( + stream_iterator: AsyncIterator[bytes], + proxy_logging_obj: ProxyLogging, + user_api_key_dict: UserAPIKeyAuth, + data: Dict[str, Any], + ): + try: + async for chunk in stream_iterator: + yield chunk + await proxy_logging_obj.update_request_status( + litellm_call_id=data.get("litellm_call_id", ""), status="success" + ) + except Exception as e: + await proxy_logging_obj.post_call_failure_hook( + user_api_key_dict=user_api_key_dict, + original_exception=e, + request_data=data, + ) + raise + finally: + if hasattr(stream_iterator, "aclose"): + await stream_iterator.aclose() # type: ignore[attr-defined] + + @staticmethod + async def get_streaming_file_content_response( + *, + custom_llm_provider: str, + file_id: str, + data: Dict[str, Any], + should_route: bool, + original_file_id: Optional[str], + credentials: Optional[Dict[str, Any]], + proxy_logging_obj: ProxyLogging, + user_api_key_dict: UserAPIKeyAuth, + version: str, + ) -> StreamingResponse: + if should_route: + prepare_data_with_credentials( + data=data, + credentials=credentials, # type: ignore[arg-type] + file_id=original_file_id, + ) + + stream_result = cast( + FileContentStreamingResult, + await litellm.afile_content( + **{ + "custom_llm_provider": custom_llm_provider, + "file_id": file_id, + "stream": True, + **data, + } # type: ignore + ), + ) + + stream_iterator = cast( + AsyncIterator[bytes], + stream_result.stream_iterator, + ) + hidden_params = getattr(stream_iterator, "_hidden_params", {}) or {} + response_headers = { + **stream_result.headers, + **ProxyBaseLLMRequestProcessing.get_custom_headers( + user_api_key_dict=user_api_key_dict, + model_id=hidden_params.get("model_id", "") or "", + cache_key=hidden_params.get("cache_key", "") or "", + api_base=hidden_params.get("api_base", "") or "", + version=version, + model_region=getattr(user_api_key_dict, "allowed_model_region", ""), + ), + } + + return StreamingResponse( + FileContentStreamingHandler.stream_file_content_with_logging( + stream_iterator=stream_iterator, + proxy_logging_obj=proxy_logging_obj, + user_api_key_dict=user_api_key_dict, + data=data, + ), + media_type="application/octet-stream", + headers=response_headers, + ) diff --git a/litellm/files/main.py b/litellm/files/main.py index 13abdfd92f..9bcb3976f0 100644 --- a/litellm/files/main.py +++ b/litellm/files/main.py @@ -771,8 +771,10 @@ async def afile_content( custom_llm_provider: FileContentProvider = "openai", extra_headers: Optional[Dict[str, str]] = None, extra_body: Optional[Dict[str, str]] = None, + chunk_size: int = 1024 * 1024, + stream: bool = False, **kwargs, -) -> HttpxBinaryResponseContent: +) -> Union[HttpxBinaryResponseContent, FileContentStreamingResult]: """ Async: Get file contents @@ -786,11 +788,13 @@ async def afile_content( # Use a partial function to pass your keyword arguments func = partial( file_content, - file_id, - model, - custom_llm_provider, - extra_headers, - extra_body, + file_id=file_id, + model=model, + custom_llm_provider=custom_llm_provider, + extra_headers=extra_headers, + extra_body=extra_body, + chunk_size=chunk_size, + stream=stream, **kwargs, ) @@ -815,8 +819,15 @@ def file_content( custom_llm_provider: Optional[Union[FileContentProvider, str]] = None, extra_headers: Optional[Dict[str, str]] = None, extra_body: Optional[Dict[str, str]] = None, + chunk_size: int = 1024 * 1024, + stream: bool = False, **kwargs, -) -> Union[HttpxBinaryResponseContent, Coroutine[Any, Any, HttpxBinaryResponseContent]]: +) -> Union[ + HttpxBinaryResponseContent, + FileContentStreamingResult, + Coroutine[Any, Any, HttpxBinaryResponseContent], + Coroutine[Any, Any, FileContentStreamingResult], +]: """ Returns the contents of the specified file. @@ -858,6 +869,23 @@ def file_content( _is_async = kwargs.pop("afile_content", False) is True + if stream: + return file_content_streaming( + file_id=file_id, + model=model, + custom_llm_provider=custom_llm_provider, + extra_headers=extra_headers, + extra_body=extra_body, + chunk_size=chunk_size, + optional_params=optional_params, + timeout=timeout, + logging_obj=cast( + Optional[LiteLLMLoggingObj], kwargs.get("litellm_logging_obj") + ), + _is_async=_is_async, + client=client, + ) + # Check if provider has a custom files config (e.g., Anthropic, Manus) provider_config = ProviderConfigManager.get_provider_files_config( model="", @@ -983,151 +1011,86 @@ def file_content( raise e -@client -async def afile_content_streaming( - file_id: str, - custom_llm_provider: FileContentProvider = "openai", - extra_headers: Optional[Dict[str, str]] = None, - extra_body: Optional[Dict[str, str]] = None, - chunk_size: int = 1024 * 1024, - **kwargs, -) -> FileContentStreamingResult: - """ - Async wrapper for file_content_streaming. - """ - try: - loop = asyncio.get_running_loop() - kwargs["afile_content_streaming"] = True - model = kwargs.pop("model", None) - - # Use a partial function to pass your keyword arguments - func = partial( - file_content_streaming, - file_id, - model, - custom_llm_provider, - extra_headers, - extra_body, - chunk_size, - **kwargs, - ) - - # Add the context to the function - ctx = contextvars.copy_context() - func_with_context = partial(ctx.run, func) - init_response = await loop.run_in_executor(None, func_with_context) - if asyncio.iscoroutine(init_response): - response = await init_response - else: - response = init_response # type: ignore - - return response - except Exception as e: - raise e - - -@client def file_content_streaming( + *, file_id: str, - model: Optional[str] = None, - custom_llm_provider: Optional[Union[FileContentProvider, str]] = None, - extra_headers: Optional[Dict[str, str]] = None, - extra_body: Optional[Dict[str, str]] = None, - chunk_size: int = 1024 * 1024, - **kwargs, + model: Optional[str], + custom_llm_provider: Optional[Union[FileContentProvider, str]], + extra_headers: Optional[Dict[str, str]], + extra_body: Optional[Dict[str, str]], + chunk_size: int, + optional_params: GenericLiteLLMParams, + timeout: Union[float, httpx.Timeout], + logging_obj: Optional[LiteLLMLoggingObj], + _is_async: bool, + client: Optional[Any], ) -> Union[FileContentStreamingResult, Coroutine[Any, Any, FileContentStreamingResult]]: - """ - Prototype API: Returns a byte iterator for file contents. + if logging_obj is not None: + logging_obj.model = model or "" + logging_obj.model_call_details["model"] = model or "" + logging_obj.model_call_details["custom_llm_provider"] = custom_llm_provider - Supports OpenAI-compatible providers and Azure. - """ - try: - optional_params = GenericLiteLLMParams(**kwargs) - logging_obj = cast( - Optional[LiteLLMLoggingObj], kwargs.get("litellm_logging_obj") + litellm_params = logging_obj.model_call_details.get("litellm_params", {}) or {} + if optional_params.api_base is not None: + litellm_params["api_base"] = optional_params.api_base + logging_obj.model_call_details["litellm_params"] = litellm_params + + def _wrap_streaming_result( + response: FileContentStreamingResult, + ) -> FileContentStreamingResult: + return FileContentStreamingResult( + stream_iterator=FileContentStreamingResponse( + stream_iterator=response.stream_iterator, + file_id=file_id, + model=model, + custom_llm_provider=custom_llm_provider, + logging_obj=logging_obj, + ), + headers=response.headers, ) - timeout = optional_params.timeout or kwargs.get("request_timeout", 600) or 600 - if ( - timeout is not None - and isinstance(timeout, httpx.Timeout) - and supports_httpx_timeout(cast(str, custom_llm_provider)) is False - ): - timeout = timeout.read or 600 - elif timeout is not None and not isinstance(timeout, httpx.Timeout): - timeout = float(timeout) # type: ignore - elif timeout is None: - timeout = 600.0 + response: Union[ + FileContentStreamingResult, Coroutine[Any, Any, FileContentStreamingResult] + ] = FileContentStreamingResult(stream_iterator=iter(()), headers={}) + if custom_llm_provider in OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS: + openai_creds = get_openai_credentials( + api_base=optional_params.api_base, + api_key=optional_params.api_key, + organization=optional_params.organization, + ) + response = openai_files_instance.file_content_streaming( + _is_async=_is_async, + file_content_request=FileContentRequest( + file_id=file_id, + extra_headers=extra_headers, + extra_body=extra_body, + ), + api_base=openai_creds.api_base, + api_key=openai_creds.api_key, + timeout=timeout, + max_retries=optional_params.max_retries, + organization=openai_creds.organization, + chunk_size=chunk_size, + client=client, + ) + else: + raise litellm.exceptions.BadRequestError( + message="LiteLLM doesn't support {} for 'file_content'. Supported providers are 'openai', 'azure', 'vertex_ai', 'bedrock', 'manus', 'anthropic'.".format( + custom_llm_provider + ), + model="n/a", + llm_provider=custom_llm_provider, + response=httpx.Response( + status_code=400, + content="Unsupported provider", + request=httpx.Request(method="create_thread", url="https://github.com/BerriAI/litellm"), # type: ignore + ), + ) - _is_async = kwargs.pop("afile_content_streaming", False) is True + if asyncio.iscoroutine(response): + async def _await_and_wrap() -> FileContentStreamingResult: + return _wrap_streaming_result(await response) - if logging_obj is not None: - logging_obj.model = model or "" - logging_obj.model_call_details["model"] = model or "" - logging_obj.model_call_details["custom_llm_provider"] = custom_llm_provider + return _await_and_wrap() - litellm_params = logging_obj.model_call_details.get("litellm_params", {}) or {} - if optional_params.api_base is not None: - litellm_params["api_base"] = optional_params.api_base - logging_obj.model_call_details["litellm_params"] = litellm_params - - def _wrap_streaming_result( - response: FileContentStreamingResult, - ) -> FileContentStreamingResult: - return FileContentStreamingResult( - stream_iterator=FileContentStreamingResponse( - stream_iterator=response.stream_iterator, - file_id=file_id, - model=model, - custom_llm_provider=custom_llm_provider, - logging_obj=logging_obj, - ), - headers=response.headers, - ) - - response: Union[ - FileContentStreamingResult, Coroutine[Any, Any, FileContentStreamingResult] - ] = FileContentStreamingResult(stream_iterator=iter(()), headers={}) - if custom_llm_provider in OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS: - openai_creds = get_openai_credentials( - api_base=optional_params.api_base, - api_key=optional_params.api_key, - organization=optional_params.organization, - ) - response = openai_files_instance.file_content_streaming( - _is_async=_is_async, - file_content_request=FileContentRequest( - file_id=file_id, - extra_headers=extra_headers, - extra_body=extra_body, - ), - api_base=openai_creds.api_base, - api_key=openai_creds.api_key, - timeout=timeout, - max_retries=optional_params.max_retries, - organization=openai_creds.organization, - chunk_size=chunk_size, - ) - else: - raise litellm.exceptions.BadRequestError( - message="LiteLLM doesn't support {} for 'file_content'. Supported providers are 'openai', 'azure', 'vertex_ai', 'bedrock', 'manus', 'anthropic'.".format( - custom_llm_provider - ), - model="n/a", - llm_provider=custom_llm_provider, - response=httpx.Response( - status_code=400, - content="Unsupported provider", - request=httpx.Request(method="create_thread", url="https://github.com/BerriAI/litellm"), # type: ignore - ), - ) - - if asyncio.iscoroutine(response): - async def _await_and_wrap() -> FileContentStreamingResult: - return _wrap_streaming_result(await response) - - return _await_and_wrap() - - return _wrap_streaming_result(response) - except Exception as e: - raise e \ No newline at end of file + return _wrap_streaming_result(response) \ No newline at end of file diff --git a/litellm/files/streaming.py b/litellm/files/streaming.py index dab190b9b3..f49b705488 100644 --- a/litellm/files/streaming.py +++ b/litellm/files/streaming.py @@ -43,6 +43,7 @@ class FileContentStreamingResponse: if logging_obj is not None and getattr(logging_obj, "start_time", None) else datetime.datetime.now() ) + self._sync_hidden_params() def __iter__(self) -> "FileContentStreamingResponse": if not hasattr(self.stream_iterator, "__next__"): diff --git a/litellm/llms/openai/openai.py b/litellm/llms/openai/openai.py index 76f300ec9c..b48edf53d5 100644 --- a/litellm/llms/openai/openai.py +++ b/litellm/llms/openai/openai.py @@ -1765,11 +1765,18 @@ class OpenAIFilesAPI(BaseLLM): headers = dict(response.headers) async def _stream() -> AsyncIterator[bytes]: + exc: Optional[BaseException] = None try: async for chunk in response.iter_bytes(chunk_size=chunk_size): yield chunk + except BaseException as e: + exc = e + raise finally: - await response_cm.__aexit__(None, None, None) + if exc is None: + await response_cm.__aexit__(None, None, None) + else: + await response_cm.__aexit__(type(exc), exc, exc.__traceback__) return FileContentStreamingResult(stream_iterator=_stream(), headers=headers) @@ -1817,10 +1824,17 @@ class OpenAIFilesAPI(BaseLLM): headers = dict(response.headers) def _stream() -> Iterator[bytes]: + exc: Optional[BaseException] = None try: yield from response.iter_bytes(chunk_size=chunk_size) + except BaseException as e: + exc = e + raise finally: - response_cm.__exit__(None, None, None) + if exc is None: + response_cm.__exit__(None, None, None) + else: + response_cm.__exit__(type(exc), exc, exc.__traceback__) return FileContentStreamingResult(stream_iterator=_stream(), headers=headers) diff --git a/litellm/proxy/openai_files_endpoints/files_endpoints.py b/litellm/proxy/openai_files_endpoints/files_endpoints.py index 468e0baaa4..eefcd12ddb 100644 --- a/litellm/proxy/openai_files_endpoints/files_endpoints.py +++ b/litellm/proxy/openai_files_endpoints/files_endpoints.py @@ -7,7 +7,7 @@ import asyncio import traceback -from typing import Any, AsyncIterator, Optional, cast, get_args +from typing import Any, Optional, cast, get_args import httpx from fastapi import ( @@ -21,11 +21,12 @@ from fastapi import ( UploadFile, status, ) -from fastapi.responses import StreamingResponse - import litellm from litellm import CreateFileRequest, get_secret_str from litellm._logging import verbose_proxy_logger +from litellm.files.file_content_streaming_handler import ( + FileContentStreamingHandler, +) from litellm.llms.base_llm.files.transformation import BaseFileEndpoints from litellm.proxy._types import * from litellm.proxy.auth.user_api_key_auth import user_api_key_auth @@ -54,7 +55,6 @@ from .common_utils import ( extract_file_creation_params, get_credentials_for_model, handle_model_based_routing, - prepare_data_with_credentials, ) from .storage_backend_service import StorageBackendFileService @@ -63,96 +63,6 @@ router = APIRouter() files_config = None -def _should_stream_file_content( - *, - custom_llm_provider: str, - is_base64_unified_file_id: Any, -) -> bool: - return ( - custom_llm_provider == "openai" - and bool(is_base64_unified_file_id) is False - ) - - -async def _stream_file_content_with_logging( - stream_iterator: AsyncIterator[bytes], - proxy_logging_obj: ProxyLogging, - user_api_key_dict: UserAPIKeyAuth, - data: Dict[str, Any], -): - try: - async for chunk in stream_iterator: - yield chunk - await proxy_logging_obj.update_request_status( - litellm_call_id=data.get("litellm_call_id", ""), status="success" - ) - except Exception as e: - await proxy_logging_obj.post_call_failure_hook( - user_api_key_dict=user_api_key_dict, - original_exception=e, - request_data=data, - ) - raise - finally: - if hasattr(stream_iterator, "aclose"): - await stream_iterator.aclose() # type: ignore[attr-defined] - - -async def _get_streaming_file_content_response( - *, - custom_llm_provider: str, - file_id: str, - data: Dict[str, Any], - should_route: bool, - original_file_id: Optional[str], - credentials: Optional[Dict[str, Any]], - proxy_logging_obj: ProxyLogging, - user_api_key_dict: UserAPIKeyAuth, - version: str, -) -> StreamingResponse: - if should_route: - prepare_data_with_credentials( - data=data, - credentials=credentials, # type: ignore[arg-type] - file_id=original_file_id, - ) - - stream_result = await litellm.afile_content_streaming( - **{ - "custom_llm_provider": custom_llm_provider, - "file_id": file_id, - **data, - } # type: ignore - ) - stream_iterator = cast( - AsyncIterator[bytes], - stream_result.stream_iterator, - ) - hidden_params = getattr(stream_iterator, "_hidden_params", {}) or {} - response_headers = { - **stream_result.headers, - **ProxyBaseLLMRequestProcessing.get_custom_headers( - user_api_key_dict=user_api_key_dict, - model_id=hidden_params.get("model_id", "") or "", - cache_key=hidden_params.get("cache_key", "") or "", - api_base=hidden_params.get("api_base", "") or "", - version=version, - model_region=getattr(user_api_key_dict, "allowed_model_region", ""), - ), - } - - return StreamingResponse( - _stream_file_content_with_logging( - stream_iterator=stream_iterator, - proxy_logging_obj=proxy_logging_obj, - user_api_key_dict=user_api_key_dict, - data=data, - ), - media_type="application/octet-stream", - headers=response_headers, - ) - - def set_files_config(config): global files_config if config is None: @@ -822,14 +732,14 @@ async def get_file_content( # noqa: PLR0915 check_file_id_encoding=True, ) - if _should_stream_file_content( + if FileContentStreamingHandler.should_stream_file_content( custom_llm_provider=custom_llm_provider, is_base64_unified_file_id=is_base64_unified_file_id, ): verbose_proxy_logger.debug( "Routing file content request to streaming response helper" ) - return await _get_streaming_file_content_response( + return await FileContentStreamingHandler.get_streaming_file_content_response( custom_llm_provider=custom_llm_provider, file_id=file_id, data=data, diff --git a/litellm/utils.py b/litellm/utils.py index d8e14de637..939bbd91ce 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -2131,8 +2131,6 @@ def _is_async_request( _STREAMING_CALL_TYPES = frozenset( { - "afile_content_streaming", - "file_content_streaming", CallTypes.generate_content_stream, CallTypes.agenerate_content_stream, CallTypes.generate_content_stream.value, diff --git a/tests/test_litellm/llms/openai/test_openai_file_content_streaming.py b/tests/test_litellm/llms/openai/test_openai_file_content_streaming.py index 66c6964021..ac5fb91f6f 100644 --- a/tests/test_litellm/llms/openai/test_openai_file_content_streaming.py +++ b/tests/test_litellm/llms/openai/test_openai_file_content_streaming.py @@ -1,14 +1,15 @@ import pytest -from typing import AsyncIterator, cast +from typing import AsyncIterator, Iterator, cast from litellm.files import main as files_main from litellm.files.streaming import FileContentStreamingResponse from litellm.files.types import FileContentStreamingResult +from litellm.llms.openai.openai import OpenAIFilesAPI from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj @pytest.mark.asyncio -async def test_afile_content_streaming_routes_to_openai_streaming_handler( +async def test_afile_content_with_stream_routes_to_openai_streaming_handler( monkeypatch, ): captured_kwargs = {} @@ -30,13 +31,17 @@ async def test_afile_content_streaming_routes_to_openai_streaming_handler( _mock_file_content_streaming, ) - stream_result = await files_main.afile_content_streaming( + stream_result = cast( + FileContentStreamingResult, + await files_main.afile_content( file_id="file-abc123", custom_llm_provider="openai", api_key="sk-test", api_base="https://api.openai.com/v1", organization="org-123", chunk_size=8, + stream=True, + ), ) async_stream_iterator = cast(AsyncIterator[bytes], stream_result.stream_iterator) @@ -50,6 +55,7 @@ async def test_afile_content_streaming_routes_to_openai_streaming_handler( assert captured_kwargs["api_base"] == "https://api.openai.com/v1" assert captured_kwargs["organization"] == "org-123" assert captured_kwargs["chunk_size"] == 8 + assert captured_kwargs["client"] is None @pytest.mark.asyncio @@ -90,11 +96,15 @@ async def test_afile_content_streaming_builds_standard_logging_object_on_complet lambda self, result, start_time, end_time, cache_hit=None: None, ) - stream_result = await files_main.afile_content_streaming( + stream_result = cast( + FileContentStreamingResult, + await files_main.afile_content( file_id="file-abc123", custom_llm_provider="openai", api_key="sk-test", api_base="https://api.openai.com/v1", + stream=True, + ), ) async_stream_iterator = cast(AsyncIterator[bytes], stream_result.stream_iterator) @@ -103,7 +113,7 @@ async def test_afile_content_streaming_builds_standard_logging_object_on_complet assert chunks == [b"hello"] assert stream_result.headers["content-length"] == "5" assert captured_standard_logging_object is not None - assert captured_standard_logging_object["call_type"] == "afile_content_streaming" + assert captured_standard_logging_object["call_type"] == "afile_content" assert captured_standard_logging_object["custom_llm_provider"] == "openai" assert captured_standard_logging_object["response"]["id"] == "file-abc123" assert ( @@ -112,6 +122,35 @@ async def test_afile_content_streaming_builds_standard_logging_object_on_complet ) +@pytest.mark.asyncio +async def test_afile_content_streaming_shim_sets_stream_flag( + monkeypatch, +): + captured_kwargs = {} + + def _mock_file_content_streaming(**kwargs): + captured_kwargs.update(kwargs) + return FileContentStreamingResult( + stream_iterator=iter(()), + headers={}, + ) + + monkeypatch.setattr( + files_main.openai_files_instance, + "file_content_streaming", + _mock_file_content_streaming, + ) + + await files_main.afile_content( + file_id="file-abc123", + custom_llm_provider="openai", + api_key="sk-test", + stream=True, + ) + + assert captured_kwargs["_is_async"] is True + + @pytest.mark.asyncio async def test_file_content_streaming_response_aclose_closes_underlying_async_generator(): close_called = False @@ -137,3 +176,148 @@ async def test_file_content_streaming_response_aclose_closes_underlying_async_ge await stream.aclose() assert close_called is True + + +@pytest.mark.asyncio +async def test_afile_content_streaming_populates_hidden_params_before_iteration( + monkeypatch, +): + async def _mock_stream(): + yield b"hello" + + def _mock_file_content_streaming(**kwargs): + return FileContentStreamingResult( + stream_iterator=_mock_stream(), + headers={"content-length": "5"}, + ) + + monkeypatch.setattr( + files_main.openai_files_instance, + "file_content_streaming", + _mock_file_content_streaming, + ) + + stream_result = cast( + FileContentStreamingResult, + await files_main.afile_content( + file_id="file-abc123", + custom_llm_provider="openai", + api_key="sk-test", + api_base="https://api.openai.com/v1", + stream=True, + ), + ) + + stream_iterator = cast(FileContentStreamingResponse, stream_result.stream_iterator) + + assert stream_iterator._hidden_params["api_base"] == "https://api.openai.com/v1" + assert stream_iterator._hidden_params["litellm_model_name"] is None + + +@pytest.mark.asyncio +async def test_afile_content_streaming_passes_exception_to_context_manager_exit(): + class MockAsyncResponse: + headers = {"content-length": "1"} + + async def iter_bytes(self, chunk_size: int): + yield b"a" + raise RuntimeError("stream failed") + + class MockAsyncResponseContextManager: + def __init__(self): + self.exc_info = None + + async def __aenter__(self): + return MockAsyncResponse() + + async def __aexit__(self, exc_type, exc, tb): + self.exc_info = (exc_type, exc, tb) + + class MockAsyncFiles: + def __init__(self, response_cm): + self.with_streaming_response = self + self._response_cm = response_cm + + def content(self, **kwargs): + return self._response_cm + + class MockAsyncOpenAIClient: + def __init__(self, response_cm): + self.files = MockAsyncFiles(response_cm) + + response_cm = MockAsyncResponseContextManager() + api = OpenAIFilesAPI() + + stream_result = await api.afile_content_streaming( + file_content_request={"file_id": "file-abc123"}, + openai_client=MockAsyncOpenAIClient(response_cm), # type: ignore[arg-type] + chunk_size=1, + ) + stream_iterator = cast(AsyncIterator[bytes], stream_result.stream_iterator) + + assert await stream_iterator.__anext__() == b"a" + + with pytest.raises(RuntimeError, match="stream failed") as exc_info: + await stream_iterator.__anext__() + + assert response_cm.exc_info is not None + assert response_cm.exc_info[0] is RuntimeError + assert response_cm.exc_info[1] is exc_info.value + assert response_cm.exc_info[2] is not None + + +def test_file_content_streaming_passes_exception_to_context_manager_exit(): + class MockSyncResponse: + headers = {"content-length": "1"} + + def iter_bytes(self, chunk_size: int) -> Iterator[bytes]: + yield b"a" + raise RuntimeError("stream failed") + + class MockSyncResponseContextManager: + def __init__(self): + self.exc_info = None + + def __enter__(self): + return MockSyncResponse() + + def __exit__(self, exc_type, exc, tb): + self.exc_info = (exc_type, exc, tb) + + class MockSyncFiles: + def __init__(self, response_cm): + self.with_streaming_response = self + self._response_cm = response_cm + + def content(self, **kwargs): + return self._response_cm + + class MockSyncOpenAIClient: + def __init__(self, response_cm): + self.files = MockSyncFiles(response_cm) + + response_cm = MockSyncResponseContextManager() + api = OpenAIFilesAPI() + + stream_result = api.file_content_streaming( + _is_async=False, + file_content_request={"file_id": "file-abc123"}, + api_base="https://api.openai.com/v1", + api_key="sk-test", + timeout=60, + max_retries=None, + organization=None, + chunk_size=1, + client=MockSyncOpenAIClient(response_cm), # type: ignore[arg-type] + ) + stream_iterator = cast(Iterator[bytes], stream_result.stream_iterator) + + assert next(stream_iterator) == b"a" + + with pytest.raises(RuntimeError, match="stream failed") as exc_info: + next(stream_iterator) + + assert response_cm.exc_info is not None + assert response_cm.exc_info[0] is RuntimeError + assert response_cm.exc_info[1] is exc_info.value + assert response_cm.exc_info[2] is not None diff --git a/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py b/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py index 3736380124..47bcaa2dfc 100644 --- a/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py +++ b/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py @@ -14,10 +14,8 @@ sys.path.insert( import litellm from litellm import Router +from litellm.files.file_content_streaming_handler import FileContentStreamingHandler from litellm.files.types import FileContentStreamingResult -from litellm.proxy.openai_files_endpoints.files_endpoints import ( - _stream_file_content_with_logging, -) from litellm.proxy._types import LiteLLM_UserTableFiltered, UserAPIKeyAuth from litellm.proxy.hooks import get_proxy_hook from litellm.proxy.management_endpoints.internal_user_endpoints import ui_view_users @@ -100,7 +98,7 @@ async def test_stream_file_content_with_logging_closes_inner_iterator_on_early_e stream_iterator = MockStreamIterator() proxy_logging_obj = AsyncMock() - generator = _stream_file_content_with_logging( + generator = FileContentStreamingHandler.stream_file_content_with_logging( stream_iterator=stream_iterator, proxy_logging_obj=proxy_logging_obj, user_api_key_dict=AsyncMock(), @@ -1606,7 +1604,7 @@ def test_get_file_content_streams_openai_direct_path( captured_kwargs = {} - async def _mock_afile_content_streaming(**kwargs): + async def _mock_afile_content(**kwargs): captured_kwargs.update(kwargs) async def _stream(): @@ -1618,11 +1616,7 @@ def test_get_file_content_streams_openai_direct_path( headers={"content-length": "11"}, ) - async def _fail_buffered_path(*args, **kwargs): - raise AssertionError("buffered afile_content path should not be used") - - monkeypatch.setattr(litellm, "afile_content_streaming", _mock_afile_content_streaming) - monkeypatch.setattr(litellm, "afile_content", _fail_buffered_path) + monkeypatch.setattr(litellm, "afile_content", _mock_afile_content) monkeypatch.setattr( "litellm.proxy.openai_files_endpoints.files_endpoints.handle_model_based_routing", lambda **kwargs: (False, None, None, None), @@ -1648,5 +1642,6 @@ def test_get_file_content_streams_openai_direct_path( assert response.headers["content-length"] == "11" assert captured_kwargs["custom_llm_provider"] == "openai" assert captured_kwargs["file_id"] == "file-abc123" + assert captured_kwargs["stream"] is True proxy_logging_obj.update_request_status.assert_awaited_once() proxy_logging_obj.post_call_failure_hook.assert_not_called() From 4e5e7395595950355d9f1836ef9dfe92a3b52884 Mon Sep 17 00:00:00 2001 From: harish876 Date: Fri, 10 Apr 2026 23:11:54 +0000 Subject: [PATCH 08/14] resolve dependency cycle --- .../file_content_streaming_handler.py | 3 ++- litellm/proxy/openai_files_endpoints/files_endpoints.py | 8 +++++--- .../proxy/openai_files_endpoint/test_files_endpoint.py | 4 +++- 3 files changed, 10 insertions(+), 5 deletions(-) rename litellm/{files => proxy/openai_files_endpoints}/file_content_streaming_handler.py (99%) diff --git a/litellm/files/file_content_streaming_handler.py b/litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py similarity index 99% rename from litellm/files/file_content_streaming_handler.py rename to litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py index dd39e1b031..b22077e8c1 100644 --- a/litellm/files/file_content_streaming_handler.py +++ b/litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py @@ -6,10 +6,11 @@ import litellm from litellm.files.types import FileContentStreamingResult from litellm.proxy._types import UserAPIKeyAuth from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing +from litellm.proxy.utils import ProxyLogging + from litellm.proxy.openai_files_endpoints.common_utils import ( prepare_data_with_credentials, ) -from litellm.proxy.utils import ProxyLogging class FileContentStreamingHandler: diff --git a/litellm/proxy/openai_files_endpoints/files_endpoints.py b/litellm/proxy/openai_files_endpoints/files_endpoints.py index eefcd12ddb..fb131f33ec 100644 --- a/litellm/proxy/openai_files_endpoints/files_endpoints.py +++ b/litellm/proxy/openai_files_endpoints/files_endpoints.py @@ -24,7 +24,7 @@ from fastapi import ( import litellm from litellm import CreateFileRequest, get_secret_str from litellm._logging import verbose_proxy_logger -from litellm.files.file_content_streaming_handler import ( +from litellm.proxy.openai_files_endpoints.file_content_streaming_handler import ( FileContentStreamingHandler, ) from litellm.llms.base_llm.files.transformation import BaseFileEndpoints @@ -49,14 +49,16 @@ from litellm.types.llms.openai import ( OpenAIFilesPurpose, ) -from .common_utils import ( +from litellm.proxy.openai_files_endpoints.common_utils import ( _is_base64_encoded_unified_file_id, encode_file_id_with_model, extract_file_creation_params, get_credentials_for_model, handle_model_based_routing, ) -from .storage_backend_service import StorageBackendFileService +from litellm.proxy.openai_files_endpoints.storage_backend_service import ( + StorageBackendFileService, +) router = APIRouter() diff --git a/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py b/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py index 47bcaa2dfc..476b57d9f3 100644 --- a/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py +++ b/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py @@ -14,11 +14,13 @@ sys.path.insert( import litellm from litellm import Router -from litellm.files.file_content_streaming_handler import FileContentStreamingHandler from litellm.files.types import FileContentStreamingResult from litellm.proxy._types import LiteLLM_UserTableFiltered, UserAPIKeyAuth from litellm.proxy.hooks import get_proxy_hook from litellm.proxy.management_endpoints.internal_user_endpoints import ui_view_users +from litellm.proxy.openai_files_endpoints.file_content_streaming_handler import ( + FileContentStreamingHandler, +) from litellm.proxy.proxy_server import app from litellm.types.llms.openai import OpenAIFileObject From d1dda3d30b3b32e79eddb3df82f6f230164d4d2a Mon Sep 17 00:00:00 2001 From: harish876 Date: Fri, 10 Apr 2026 23:29:16 +0000 Subject: [PATCH 09/14] Enhance file content streaming handler to support custom LLM provider routing - Updated `FileContentStreamingHandler` to utilize `custom_llm_provider` from credentials for routing. - Added error handling for missing `custom_llm_provider` in credentials. - Introduced new tests to validate streaming behavior with routed providers and non-OpenAI providers. - Cleaned up imports and ensured proper type casting for improved clarity. --- .../file_content_streaming_handler.py | 6 +- .../test_files_endpoint.py | 66 +++++++++++++++++++ 2 files changed, 71 insertions(+), 1 deletion(-) diff --git a/litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py b/litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py index b22077e8c1..9833891f1d 100644 --- a/litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py +++ b/litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py @@ -62,18 +62,22 @@ class FileContentStreamingHandler: user_api_key_dict: UserAPIKeyAuth, version: str, ) -> StreamingResponse: + effective_custom_llm_provider = custom_llm_provider if should_route: prepare_data_with_credentials( data=data, credentials=credentials, # type: ignore[arg-type] file_id=original_file_id, ) + effective_custom_llm_provider = cast( + str, credentials["custom_llm_provider"] + ) stream_result = cast( FileContentStreamingResult, await litellm.afile_content( **{ - "custom_llm_provider": custom_llm_provider, + "custom_llm_provider": effective_custom_llm_provider, "file_id": file_id, "stream": True, **data, diff --git a/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py b/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py index 476b57d9f3..aeca75e936 100644 --- a/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py +++ b/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py @@ -1647,3 +1647,69 @@ def test_get_file_content_streams_openai_direct_path( assert captured_kwargs["stream"] is True proxy_logging_obj.update_request_status.assert_awaited_once() proxy_logging_obj.post_call_failure_hook.assert_not_called() + + +def test_get_file_content_streams_with_routed_provider( + mocker: MockerFixture, monkeypatch, llm_router: Router +): + import litellm.proxy.proxy_server as ps + from litellm.proxy._types import LitellmUserRoles + + proxy_logging_obj = setup_proxy_logging_object(monkeypatch, llm_router) + monkeypatch.setattr("litellm.proxy.proxy_server.master_key", None) + monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", None) + proxy_logging_obj.update_request_status = mocker.AsyncMock() + proxy_logging_obj.post_call_failure_hook = mocker.AsyncMock() + + captured_kwargs = {} + + async def _mock_afile_content(**kwargs): + captured_kwargs.update(kwargs) + + async def _stream(): + yield b"hello " + yield b"world" + + return FileContentStreamingResult( + stream_iterator=_stream(), + headers={"content-length": "11"}, + ) + + monkeypatch.setattr(litellm, "afile_content", _mock_afile_content) + monkeypatch.setattr( + "litellm.proxy.openai_files_endpoints.files_endpoints.handle_model_based_routing", + lambda **kwargs: ( + True, + "azure-gpt-3-5-turbo", + "file-original-123", + { + "custom_llm_provider": "azure", + "api_key": "azure-key", + "api_base": "https://azure.example.com", + }, + ), + ) + + app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth( + api_key="test-key", + user_role=LitellmUserRoles.PROXY_ADMIN, + user_id="test-user", + ) + + try: + response = client.get( + "/v1/files/file-abc123/content", + headers={"Authorization": "Bearer test-key"}, + ) + finally: + app.dependency_overrides.pop(ps.user_api_key_auth, None) + + assert response.status_code == 200, response.text + assert response.content == b"hello world" + assert captured_kwargs["custom_llm_provider"] == "azure" + assert captured_kwargs["file_id"] == "file-original-123" + assert captured_kwargs["api_key"] == "azure-key" + assert captured_kwargs["api_base"] == "https://azure.example.com" + assert captured_kwargs["stream"] is True + proxy_logging_obj.update_request_status.assert_awaited_once() + proxy_logging_obj.post_call_failure_hook.assert_not_called() From c5d93e67f4becf4f65ace595265ed9787656f378 Mon Sep 17 00:00:00 2001 From: harish876 Date: Fri, 10 Apr 2026 23:29:44 +0000 Subject: [PATCH 10/14] Enhance error handling in FileContentStreamingHandler for custom LLM provider routing - Added validation to ensure credentials include a custom LLM provider before routing. - Cleaned up type casting for better readability. - Introduced a new test to verify behavior when a non-OpenAI provider is used, ensuring proper handling of streaming responses. - Updated imports to include necessary modules for testing. --- .../file_content_streaming_handler.py | 10 +-- .../test_files_endpoint.py | 67 ++++++++++++++++++- 2 files changed, 72 insertions(+), 5 deletions(-) diff --git a/litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py b/litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py index 9833891f1d..e992c47f28 100644 --- a/litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py +++ b/litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py @@ -64,14 +64,16 @@ class FileContentStreamingHandler: ) -> StreamingResponse: effective_custom_llm_provider = custom_llm_provider if should_route: + if credentials is None or credentials.get("custom_llm_provider") is None: + raise ValueError( + "Model-based file routing requires credentials with custom_llm_provider" + ) prepare_data_with_credentials( data=data, - credentials=credentials, # type: ignore[arg-type] + credentials=credentials, file_id=original_file_id, ) - effective_custom_llm_provider = cast( - str, credentials["custom_llm_provider"] - ) + effective_custom_llm_provider = cast(str, credentials["custom_llm_provider"]) stream_result = cast( FileContentStreamingResult, diff --git a/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py b/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py index aeca75e936..66afded70a 100644 --- a/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py +++ b/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py @@ -5,6 +5,7 @@ from unittest.mock import ANY, AsyncMock import pytest import respx +import httpx from fastapi.testclient import TestClient from pytest_mock import MockerFixture @@ -22,7 +23,7 @@ from litellm.proxy.openai_files_endpoints.file_content_streaming_handler import FileContentStreamingHandler, ) from litellm.proxy.proxy_server import app -from litellm.types.llms.openai import OpenAIFileObject +from litellm.types.llms.openai import HttpxBinaryResponseContent, OpenAIFileObject client = TestClient(app) from litellm.caching.caching import DualCache @@ -1713,3 +1714,67 @@ def test_get_file_content_streams_with_routed_provider( assert captured_kwargs["stream"] is True proxy_logging_obj.update_request_status.assert_awaited_once() proxy_logging_obj.post_call_failure_hook.assert_not_called() + + +def test_get_file_content_non_openai_provider_skips_streaming_handler( + mocker: MockerFixture, monkeypatch, llm_router: Router +): + import litellm.proxy.proxy_server as ps + from litellm.proxy._types import LitellmUserRoles + + proxy_logging_obj = setup_proxy_logging_object(monkeypatch, llm_router) + monkeypatch.setattr("litellm.proxy.proxy_server.master_key", None) + monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", None) + proxy_logging_obj.post_call_failure_hook = mocker.AsyncMock() + + captured_kwargs = {} + + async def _mock_afile_content(**kwargs): + captured_kwargs.update(kwargs) + return HttpxBinaryResponseContent( + response=httpx.Response( + status_code=200, + content=b"azure-bytes", + headers={ + "content-type": "application/octet-stream", + "content-length": "11", + }, + ) + ) + + mock_streaming_response = mocker.AsyncMock() + + monkeypatch.setattr(litellm, "afile_content", _mock_afile_content) + monkeypatch.setattr( + FileContentStreamingHandler, + "get_streaming_file_content_response", + mock_streaming_response, + ) + monkeypatch.setattr( + "litellm.proxy.openai_files_endpoints.files_endpoints.handle_model_based_routing", + lambda **kwargs: (False, None, None, None), + ) + + app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth( + api_key="test-key", + user_role=LitellmUserRoles.PROXY_ADMIN, + user_id="test-user", + ) + + try: + response = client.get( + "/v1/files/file-abc123/content", + headers={ + "Authorization": "Bearer test-key", + "custom-llm-provider": "azure", + }, + ) + finally: + app.dependency_overrides.pop(ps.user_api_key_auth, None) + + assert response.status_code == 200, response.text + assert response.content == b"azure-bytes" + assert captured_kwargs["custom_llm_provider"] == "azure" + assert "stream" not in captured_kwargs + mock_streaming_response.assert_not_awaited() + proxy_logging_obj.post_call_failure_hook.assert_not_called() From 68bc6de214cbe0346849e3319f12eabbedb3d963 Mon Sep 17 00:00:00 2001 From: harish876 Date: Sat, 11 Apr 2026 00:07:41 +0000 Subject: [PATCH 11/14] resolving dependency issues --- .../file_content_streaming_handler.py | 26 +++++++++++-------- .../openai_files_endpoints/files_endpoints.py | 14 +++++----- 2 files changed, 23 insertions(+), 17 deletions(-) diff --git a/litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py b/litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py index e992c47f28..2ca39591dc 100644 --- a/litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py +++ b/litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py @@ -1,16 +1,13 @@ -from typing import Any, AsyncIterator, Dict, Optional, cast +from typing import TYPE_CHECKING, Any, AsyncIterator, Dict, Optional, cast from fastapi.responses import StreamingResponse import litellm from litellm.files.types import FileContentStreamingResult -from litellm.proxy._types import UserAPIKeyAuth -from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing -from litellm.proxy.utils import ProxyLogging -from litellm.proxy.openai_files_endpoints.common_utils import ( - prepare_data_with_credentials, -) +if TYPE_CHECKING: + from litellm.proxy._types import UserAPIKeyAuth + from litellm.proxy.utils import ProxyLogging class FileContentStreamingHandler: @@ -28,8 +25,8 @@ class FileContentStreamingHandler: @staticmethod async def stream_file_content_with_logging( stream_iterator: AsyncIterator[bytes], - proxy_logging_obj: ProxyLogging, - user_api_key_dict: UserAPIKeyAuth, + proxy_logging_obj: "ProxyLogging", + user_api_key_dict: "UserAPIKeyAuth", data: Dict[str, Any], ): try: @@ -58,10 +55,17 @@ class FileContentStreamingHandler: should_route: bool, original_file_id: Optional[str], credentials: Optional[Dict[str, Any]], - proxy_logging_obj: ProxyLogging, - user_api_key_dict: UserAPIKeyAuth, + proxy_logging_obj: "ProxyLogging", + user_api_key_dict: "UserAPIKeyAuth", version: str, ) -> StreamingResponse: + from litellm.proxy.common_request_processing import ( + ProxyBaseLLMRequestProcessing, + ) + from litellm.proxy.openai_files_endpoints.common_utils import ( + prepare_data_with_credentials, + ) + effective_custom_llm_provider = custom_llm_provider if should_route: if credentials is None or credentials.get("custom_llm_provider") is None: diff --git a/litellm/proxy/openai_files_endpoints/files_endpoints.py b/litellm/proxy/openai_files_endpoints/files_endpoints.py index fb131f33ec..cdf4785f98 100644 --- a/litellm/proxy/openai_files_endpoints/files_endpoints.py +++ b/litellm/proxy/openai_files_endpoints/files_endpoints.py @@ -24,9 +24,6 @@ from fastapi import ( import litellm from litellm import CreateFileRequest, get_secret_str from litellm._logging import verbose_proxy_logger -from litellm.proxy.openai_files_endpoints.file_content_streaming_handler import ( - FileContentStreamingHandler, -) from litellm.llms.base_llm.files.transformation import BaseFileEndpoints from litellm.proxy._types import * from litellm.proxy.auth.user_api_key_auth import user_api_key_auth @@ -55,9 +52,7 @@ from litellm.proxy.openai_files_endpoints.common_utils import ( extract_file_creation_params, get_credentials_for_model, handle_model_based_routing, -) -from litellm.proxy.openai_files_endpoints.storage_backend_service import ( - StorageBackendFileService, + prepare_data_with_credentials, ) router = APIRouter() @@ -162,6 +157,9 @@ async def route_create_file( from litellm.litellm_core_utils.prompt_templates.common_utils import ( extract_file_data, ) + from litellm.proxy.openai_files_endpoints.storage_backend_service import ( + StorageBackendFileService, + ) # Extract file data file_data = extract_file_data(cast(Any, _create_file_request.get("file"))) @@ -734,6 +732,10 @@ async def get_file_content( # noqa: PLR0915 check_file_id_encoding=True, ) + from litellm.proxy.openai_files_endpoints.file_content_streaming_handler import ( + FileContentStreamingHandler, + ) + if FileContentStreamingHandler.should_stream_file_content( custom_llm_provider=custom_llm_provider, is_base64_unified_file_id=is_base64_unified_file_id, From f523ccb2fe63016ebfc7dae278c08cc0e772fe06 Mon Sep 17 00:00:00 2001 From: harish876 Date: Sat, 11 Apr 2026 00:07:52 +0000 Subject: [PATCH 12/14] Update import paths in tests for StorageBackendFileService - Changed the import path for `upload_file_to_storage_backend` in test files to reflect the new module structure. - Ensured consistency in mocking for storage backend service tests. --- .../proxy/openai_files_endpoint/test_files_endpoint.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py b/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py index 66afded70a..1196944f3a 100644 --- a/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py +++ b/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py @@ -322,7 +322,7 @@ def test_target_storage_invokes_storage_backend( ) ) mocker.patch( - "litellm.proxy.openai_files_endpoints.files_endpoints.StorageBackendFileService.upload_file_to_storage_backend", + "litellm.proxy.openai_files_endpoints.storage_backend_service.StorageBackendFileService.upload_file_to_storage_backend", new=async_mock, ) @@ -381,7 +381,7 @@ def test_target_storage_with_target_models( ) ) mocker.patch( - "litellm.proxy.openai_files_endpoints.files_endpoints.StorageBackendFileService.upload_file_to_storage_backend", + "litellm.proxy.openai_files_endpoints.storage_backend_service.StorageBackendFileService.upload_file_to_storage_backend", new=async_mock, ) From 69eb34597cad1c04c5d124c6723ca827ccaa45ad Mon Sep 17 00:00:00 2001 From: harish876 Date: Sat, 11 Apr 2026 18:56:15 +0000 Subject: [PATCH 13/14] Refactor file content streaming handling to improve routing and support - Introduced a new method in `FileContentStreamingHandler` to resolve streaming request parameters, enhancing the routing logic based on credentials. - Updated the `should_stream_file_content` method to check against supported providers. - Cleaned up type hints and imports across multiple files for better organization and clarity. - Added comprehensive tests to validate the new routing behavior and ensure original data integrity during streaming requests. --- litellm/files/main.py | 22 ++-- litellm/files/streaming.py | 8 +- litellm/files/types.py | 7 +- .../file_content_streaming_handler.py | 81 ++++++++---- .../openai_files_endpoints/files_endpoints.py | 33 +++-- .../test_files_endpoint.py | 119 ++++++++++++++++-- 6 files changed, 206 insertions(+), 64 deletions(-) diff --git a/litellm/files/main.py b/litellm/files/main.py index 9bcb3976f0..46199a4eca 100644 --- a/litellm/files/main.py +++ b/litellm/files/main.py @@ -30,14 +30,10 @@ FileRetrieveProvider = Literal[ ] FileDeleteProvider = Literal["openai", "azure", "gemini", "manus", "anthropic"] FileListProvider = Literal["openai", "azure", "manus", "anthropic"] -FileContentProvider = Literal[ - "openai", "azure", "vertex_ai", "bedrock", "hosted_vllm", "anthropic", "manus" -] - import litellm from litellm import get_secret_str from litellm.files.streaming import FileContentStreamingResponse -from litellm.files.types import FileContentStreamingResult +from litellm.files.types import FileContentProvider, FileContentStreamingResult from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.llms.azure.common_utils import get_azure_credentials @@ -68,6 +64,15 @@ from litellm.utils import ( base_llm_http_handler = BaseLLMHTTPHandler() ####### ENVIRONMENT VARIABLES ################### + + +def _should_sdk_support_streaming( + custom_llm_provider: Optional[Union[FileContentProvider, str]], +) -> bool: + """ + Return whether file content streaming is supported for the provider. + """ + return custom_llm_provider in OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS openai_files_instance = OpenAIFilesAPI() azure_files_instance = AzureOpenAIFilesAPI() vertex_ai_files_instance = VertexAIFilesHandler() @@ -869,7 +874,7 @@ def file_content( _is_async = kwargs.pop("afile_content", False) is True - if stream: + if stream and _should_sdk_support_streaming(custom_llm_provider): return file_content_streaming( file_id=file_id, model=model, @@ -1075,8 +1080,9 @@ def file_content_streaming( ) else: raise litellm.exceptions.BadRequestError( - message="LiteLLM doesn't support {} for 'file_content'. Supported providers are 'openai', 'azure', 'vertex_ai', 'bedrock', 'manus', 'anthropic'.".format( - custom_llm_provider + message="LiteLLM doesn't support {} for streaming 'file_content'. Supported providers are {}.".format( + custom_llm_provider, + sorted(OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS), ), model="n/a", llm_provider=custom_llm_provider, diff --git a/litellm/files/streaming.py b/litellm/files/streaming.py index f49b705488..a92b92abb2 100644 --- a/litellm/files/streaming.py +++ b/litellm/files/streaming.py @@ -1,8 +1,9 @@ import datetime import traceback -from typing import TYPE_CHECKING, Any, AsyncIterator, Dict, Iterator, Literal, Optional, Union, cast +from typing import TYPE_CHECKING, Any, AsyncIterator, Dict, Iterator, Optional, Union, cast import anyio +from litellm.files.types import FileContentProvider if TYPE_CHECKING: from litellm.litellm_core_utils.litellm_logging import ( @@ -10,11 +11,6 @@ if TYPE_CHECKING: ) from litellm.types.utils import StandardLoggingHiddenParams, StandardLoggingPayload -FileContentProvider = Literal[ - "openai", "azure", "vertex_ai", "bedrock", "hosted_vllm", "anthropic", "manus" -] - - class FileContentStreamingResponse: """ Iterator wrapper for file content streaming that carries LiteLLM metadata diff --git a/litellm/files/types.py b/litellm/files/types.py index 2357bd997f..688bc86f0c 100644 --- a/litellm/files/types.py +++ b/litellm/files/types.py @@ -1,4 +1,9 @@ -from typing import AsyncIterator, Dict, Iterator, NamedTuple, Union +from typing import AsyncIterator, Dict, Iterator, Literal, NamedTuple, Union + + +FileContentProvider = Literal[ + "openai", "azure", "vertex_ai", "bedrock", "hosted_vllm", "anthropic", "manus" +] class FileContentStreamingResult(NamedTuple): diff --git a/litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py b/litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py index 2ca39591dc..7ccec3a748 100644 --- a/litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py +++ b/litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py @@ -1,9 +1,10 @@ -from typing import TYPE_CHECKING, Any, AsyncIterator, Dict, Optional, cast +from typing import TYPE_CHECKING, Any, AsyncIterator, Dict, Optional, Tuple, cast from fastapi.responses import StreamingResponse import litellm -from litellm.files.types import FileContentStreamingResult +from litellm.files.types import FileContentProvider, FileContentStreamingResult +from litellm.types.utils import OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS if TYPE_CHECKING: from litellm.proxy._types import UserAPIKeyAuth @@ -11,15 +12,61 @@ if TYPE_CHECKING: class FileContentStreamingHandler: + @staticmethod + def resolve_streaming_request_params( + *, + custom_llm_provider: str, + file_id: str, + data: Dict[str, Any], + should_route: bool, + original_file_id: Optional[str], + credentials: Optional[Dict[str, Any]], + ) -> Tuple[str, str, Dict[str, Any]]: + """ + Resolve the provider, file ID, and request payload to use for streaming. + + For model-routed requests, this derives the effective provider from + credentials, applies `prepare_data_with_credentials()` to a copied + payload, swaps in the decoded/original file ID, and removes `model` + so `afile_content()` does not re-resolve the provider. This helper + does not mutate the passed-in `data` dictionary. Non-routed requests + return the original provider, file ID, and data unchanged. + """ + if should_route and credentials is not None: + from litellm.proxy.openai_files_endpoints.common_utils import ( + prepare_data_with_credentials, + ) + + resolved_streaming_data = dict(data) + prepare_data_with_credentials( + data=resolved_streaming_data, + credentials=credentials, + file_id=original_file_id, + ) + resolved_streaming_data.pop("model", None) + resolved_streaming_provider = cast( + str, credentials["custom_llm_provider"] + ) + resolved_custom_llm_provider = resolved_streaming_provider + resolved_file_id = cast(str, resolved_streaming_data["file_id"]) + else: + resolved_streaming_data = data + resolved_custom_llm_provider = custom_llm_provider + resolved_file_id = file_id + + return ( + resolved_custom_llm_provider, + resolved_file_id, + resolved_streaming_data, + ) + @staticmethod def should_stream_file_content( *, custom_llm_provider: str, - is_base64_unified_file_id: Any, ) -> bool: return ( - custom_llm_provider == "openai" - and bool(is_base64_unified_file_id) is False + custom_llm_provider in OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS ) @staticmethod @@ -52,9 +99,6 @@ class FileContentStreamingHandler: custom_llm_provider: str, file_id: str, data: Dict[str, Any], - should_route: bool, - original_file_id: Optional[str], - credentials: Optional[Dict[str, Any]], proxy_logging_obj: "ProxyLogging", user_api_key_dict: "UserAPIKeyAuth", version: str, @@ -62,28 +106,13 @@ class FileContentStreamingHandler: from litellm.proxy.common_request_processing import ( ProxyBaseLLMRequestProcessing, ) - from litellm.proxy.openai_files_endpoints.common_utils import ( - prepare_data_with_credentials, - ) - - effective_custom_llm_provider = custom_llm_provider - if should_route: - if credentials is None or credentials.get("custom_llm_provider") is None: - raise ValueError( - "Model-based file routing requires credentials with custom_llm_provider" - ) - prepare_data_with_credentials( - data=data, - credentials=credentials, - file_id=original_file_id, - ) - effective_custom_llm_provider = cast(str, credentials["custom_llm_provider"]) - stream_result = cast( FileContentStreamingResult, await litellm.afile_content( **{ - "custom_llm_provider": effective_custom_llm_provider, + "custom_llm_provider": cast( + FileContentProvider, custom_llm_provider + ), "file_id": file_id, "stream": True, **data, diff --git a/litellm/proxy/openai_files_endpoints/files_endpoints.py b/litellm/proxy/openai_files_endpoints/files_endpoints.py index cdf4785f98..b327d2be73 100644 --- a/litellm/proxy/openai_files_endpoints/files_endpoints.py +++ b/litellm/proxy/openai_files_endpoints/files_endpoints.py @@ -634,7 +634,7 @@ async def get_file_content( # noqa: PLR0915 or await get_custom_llm_provider_from_request_body(request=request) or "openai" ) - + ## check if file_id is a litellm managed file is_base64_unified_file_id = _is_base64_encoded_unified_file_id(file_id) if is_base64_unified_file_id: @@ -735,21 +735,33 @@ async def get_file_content( # noqa: PLR0915 from litellm.proxy.openai_files_endpoints.file_content_streaming_handler import ( FileContentStreamingHandler, ) + ( + resolved_custom_llm_provider, + resolved_file_id, + resolved_streaming_data, + ) = FileContentStreamingHandler.resolve_streaming_request_params( + custom_llm_provider=custom_llm_provider, + file_id=file_id, + data=data, + should_route=should_route, + original_file_id=original_file_id, + credentials=credentials, + ) if FileContentStreamingHandler.should_stream_file_content( - custom_llm_provider=custom_llm_provider, - is_base64_unified_file_id=is_base64_unified_file_id, + custom_llm_provider=resolved_custom_llm_provider, ): verbose_proxy_logger.debug( - "Routing file content request to streaming response helper" + "Using streaming file content helper for custom_llm_provider=%s, original_file_id=%s, file_id=%s, model_used=%s", + resolved_custom_llm_provider, + original_file_id, + resolved_file_id, + model_used, ) return await FileContentStreamingHandler.get_streaming_file_content_response( - custom_llm_provider=custom_llm_provider, - file_id=file_id, - data=data, - should_route=should_route, - original_file_id=original_file_id, - credentials=credentials, + custom_llm_provider=resolved_custom_llm_provider, + file_id=resolved_file_id, + data=resolved_streaming_data, proxy_logging_obj=proxy_logging_obj, user_api_key_dict=user_api_key_dict, version=version, @@ -762,7 +774,6 @@ async def get_file_content( # noqa: PLR0915 credentials=credentials, # type: ignore file_id=original_file_id, # Use decoded file ID if from encoded ID ) - response = await litellm.afile_content( custom_llm_provider=credentials["custom_llm_provider"], # type: ignore **data, diff --git a/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py b/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py index 1196944f3a..09d11388d8 100644 --- a/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py +++ b/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py @@ -116,6 +116,93 @@ async def test_stream_file_content_with_logging_closes_inner_iterator_on_early_e proxy_logging_obj.update_request_status.assert_not_called() +def test_resolve_streaming_request_params_non_routed_returns_original_values(): + data = {"file_id": "file-abc123", "metadata": {"k": "v"}} + + ( + resolved_custom_llm_provider, + resolved_file_id, + resolved_streaming_data, + ) = FileContentStreamingHandler.resolve_streaming_request_params( + custom_llm_provider="openai", + file_id="file-abc123", + data=data, + should_route=False, + original_file_id=None, + credentials=None, + ) + + assert resolved_custom_llm_provider == "openai" + assert resolved_file_id == "file-abc123" + assert resolved_streaming_data is data + + +def test_resolve_streaming_request_params_routed_uses_credentials_and_original_file_id(): + data = { + "file_id": "file-encoded-123", + "model": "azure-gpt-3-5-turbo", + "metadata": {"k": "v"}, + } + credentials = { + "custom_llm_provider": "azure", + "api_key": "azure-key", + "api_base": "https://azure.example.com", + } + + ( + resolved_custom_llm_provider, + resolved_file_id, + resolved_streaming_data, + ) = FileContentStreamingHandler.resolve_streaming_request_params( + custom_llm_provider="openai", + file_id="file-encoded-123", + data=data, + should_route=True, + original_file_id="file-original-123", + credentials=credentials, + ) + + assert resolved_custom_llm_provider == "azure" + assert resolved_file_id == "file-original-123" + assert resolved_streaming_data["file_id"] == "file-original-123" + assert resolved_streaming_data["api_key"] == "azure-key" + assert resolved_streaming_data["api_base"] == "https://azure.example.com" + assert "custom_llm_provider" not in resolved_streaming_data + assert "model" not in resolved_streaming_data + assert data["file_id"] == "file-encoded-123" + assert data["model"] == "azure-gpt-3-5-turbo" + + +def test_resolve_streaming_request_params_routed_preserves_input_data_object(): + data = { + "file_id": "file-encoded-123", + "model": "openai/gpt-4o", + } + credentials = { + "custom_llm_provider": "openai", + "api_key": "sk-test", + } + + ( + _resolved_custom_llm_provider, + _resolved_file_id, + resolved_streaming_data, + ) = FileContentStreamingHandler.resolve_streaming_request_params( + custom_llm_provider="openai", + file_id="file-encoded-123", + data=data, + should_route=True, + original_file_id=None, + credentials=credentials, + ) + + assert resolved_streaming_data is not data + assert data == { + "file_id": "file-encoded-123", + "model": "openai/gpt-4o", + } + + def test_invalid_purpose(mocker: MockerFixture, monkeypatch, llm_router: Router): """ Asserts 'create_file' is called with the correct arguments @@ -1650,7 +1737,7 @@ def test_get_file_content_streams_openai_direct_path( proxy_logging_obj.post_call_failure_hook.assert_not_called() -def test_get_file_content_streams_with_routed_provider( +def test_get_file_content_routed_provider_skips_streaming_when_resolved_provider_is_not_supported( mocker: MockerFixture, monkeypatch, llm_router: Router ): import litellm.proxy.proxy_server as ps @@ -1666,17 +1753,25 @@ def test_get_file_content_streams_with_routed_provider( async def _mock_afile_content(**kwargs): captured_kwargs.update(kwargs) - - async def _stream(): - yield b"hello " - yield b"world" - - return FileContentStreamingResult( - stream_iterator=_stream(), - headers={"content-length": "11"}, + return HttpxBinaryResponseContent( + response=httpx.Response( + status_code=200, + content=b"azure-bytes", + headers={ + "content-type": "application/octet-stream", + "content-length": "11", + }, + ) ) + mock_streaming_response = mocker.AsyncMock() + monkeypatch.setattr(litellm, "afile_content", _mock_afile_content) + monkeypatch.setattr( + FileContentStreamingHandler, + "get_streaming_file_content_response", + mock_streaming_response, + ) monkeypatch.setattr( "litellm.proxy.openai_files_endpoints.files_endpoints.handle_model_based_routing", lambda **kwargs: ( @@ -1706,13 +1801,13 @@ def test_get_file_content_streams_with_routed_provider( app.dependency_overrides.pop(ps.user_api_key_auth, None) assert response.status_code == 200, response.text - assert response.content == b"hello world" + assert response.content == b"azure-bytes" assert captured_kwargs["custom_llm_provider"] == "azure" assert captured_kwargs["file_id"] == "file-original-123" assert captured_kwargs["api_key"] == "azure-key" assert captured_kwargs["api_base"] == "https://azure.example.com" - assert captured_kwargs["stream"] is True - proxy_logging_obj.update_request_status.assert_awaited_once() + assert "stream" not in captured_kwargs + mock_streaming_response.assert_not_awaited() proxy_logging_obj.post_call_failure_hook.assert_not_called() From ec0cd5c17d810a2f063c2d286a29f4c4fd5a9050 Mon Sep 17 00:00:00 2001 From: harish-berri Date: Sat, 11 Apr 2026 13:04:25 -0700 Subject: [PATCH 14/14] Update streaming.py. Provide Type Annotation for empty dict --- litellm/files/streaming.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/litellm/files/streaming.py b/litellm/files/streaming.py index a92b92abb2..36fe30fa82 100644 --- a/litellm/files/streaming.py +++ b/litellm/files/streaming.py @@ -118,7 +118,7 @@ class FileContentStreamingResponse: return response def _sync_hidden_params(self) -> None: - litellm_params = {} + litellm_params: dict[str, Any] = {} if self.logging_obj is not None: litellm_params = ( self.logging_obj.model_call_details.get("litellm_params", {}) or {}