diff --git a/litellm/constants.py b/litellm/constants.py index da731cb5eb2..2ee0e36be75 100644 --- a/litellm/constants.py +++ b/litellm/constants.py @@ -53,6 +53,7 @@ S3_BOUNDED_OBJECT_KEY_HEAD_BYTES: Final = 64 S3_PREFIX_DIGEST_CHARS: Final = 16 # s3 allows 2048 bytes of combined metadata headers, which Content-Disposition counts against MAX_S3_OBJECT_DOWNLOAD_FILENAME_BYTES: Final = 1024 +MAX_FILE_LIST_LIMIT: Final = 10000 DEFAULT_SQS_FLUSH_INTERVAL_SECONDS: Final = int(os.getenv("DEFAULT_SQS_FLUSH_INTERVAL_SECONDS", 10)) DEFAULT_NUM_WORKERS_LITELLM_PROXY: Final = int(os.getenv("DEFAULT_NUM_WORKERS_LITELLM_PROXY", 1)) DYNAMIC_RATE_LIMIT_ERROR_THRESHOLD_PER_MINUTE = int(os.getenv("DYNAMIC_RATE_LIMIT_ERROR_THRESHOLD_PER_MINUTE", 1)) diff --git a/litellm/llms/custom_httpx/llm_http_handler.py b/litellm/llms/custom_httpx/llm_http_handler.py index e5389d0e0b7..a70807bf6a5 100644 --- a/litellm/llms/custom_httpx/llm_http_handler.py +++ b/litellm/llms/custom_httpx/llm_http_handler.py @@ -18,7 +18,7 @@ import litellm.types import litellm.types.utils from litellm._logging import _redact_string, verbose_logger from litellm.anthropic_beta_headers_manager import update_headers_with_filtered_beta -from litellm.constants import REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES +from litellm.constants import MAX_FILE_LIST_LIMIT, REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES from litellm.litellm_core_utils.agentic_loop_settings import ( DEFAULT_MAX_AGENTIC_LOOPS, validated_max_agentic_loops, @@ -4924,15 +4924,16 @@ class BaseLLMHTTPHandler: ) try: - response: Final = sync_httpx_client.get(url=url, headers=headers, params=params) + response: Final = sync_httpx_client.get(url=url, headers=headers, params=params, timeout=timeout) except Exception as e: raise self._handle_error(e=e, provider_config=provider_config) - pages: Final = ( - response, - *self._following_list_files_pages(response, provider_config, litellm_params, headers, sync_httpx_client), + files_per_page: Final = self._files_per_listing_page( + response, provider_config, logging_obj, litellm_params, headers, sync_httpx_client, timeout ) - return self._listed_files_across_pages(pages, provider_config, logging_obj, litellm_params) + return [ # mutable-ok: the base files contract returns a list + listed_file for page_files in files_per_page for listed_file in page_files + ] async def async_list_files( self, @@ -4980,48 +4981,38 @@ class BaseLLMHTTPHandler: ) try: - response: Final = await async_httpx_client.get(url=url, headers=headers, params=params) + response: Final = await async_httpx_client.get(url=url, headers=headers, params=params, timeout=timeout) except Exception as e: raise self._handle_error(e=e, provider_config=provider_config) - following_pages: Final = self._following_async_list_files_pages( - response, provider_config, litellm_params, headers, async_httpx_client + files_per_page: Final = self._files_per_async_listing_page( + response, provider_config, logging_obj, litellm_params, headers, async_httpx_client, timeout ) - pages: Final = ( - response, - *[page async for page in following_pages], # mutable-ok: an async comprehension is spelled as a list - ) - return self._listed_files_across_pages(pages, provider_config, logging_obj, litellm_params) + return [ # mutable-ok: the base files contract returns a list + listed_file async for page_files in files_per_page for listed_file in page_files + ] - def _listed_files_across_pages( + def _files_per_listing_page( self, - pages: Sequence[httpx.Response], + first_page: httpx.Response, provider_config: BaseFilesConfig, logging_obj: LiteLLMLoggingObj, litellm_params: dict, # mutable-ok: handed to the files contract, which types it as a dict - ) -> list[OpenAIFileObject]: - return [ # mutable-ok: the base files contract returns a list - listed_file - for page in pages - for listed_file in provider_config.transform_list_files_response( - raw_response=page, - logging_obj=logging_obj, - litellm_params=litellm_params, - ) - ] - - def _following_list_files_pages( - self, - page: httpx.Response, - provider_config: BaseFilesConfig, - litellm_params: dict, # mutable-ok: handed to the files contract, which types it as a dict headers: dict, # mutable-ok: handed to validate_environment, which types it as a dict client: HTTPHandler, - ) -> Iterator[httpx.Response]: - latest_page = page # rebind-ok: advances one page per loop turn - while next_request := provider_config.transform_list_files_next_request( - raw_response=latest_page, optional_params={}, litellm_params=litellm_params - ): + timeout: float | httpx.Timeout | None, + ) -> Iterator[list[OpenAIFileObject]]: # mutable-ok: each page arrives as the list the files contract returns + latest_page = first_page # rebind-ok: advances one page per loop turn + listed_count = 0 # rebind-ok: grows per page so the listing stops at MAX_FILE_LIST_LIMIT, OpenAI's ceiling + while True: + page_files = provider_config.transform_list_files_response( + raw_response=latest_page, logging_obj=logging_obj, litellm_params=litellm_params + ) + yield page_files[: MAX_FILE_LIST_LIMIT - listed_count] + listed_count += len(page_files) + next_request = self._next_listing_request(latest_page, provider_config, litellm_params, listed_count) + if next_request is None: + return url, params = next_request next_headers = provider_config.validate_environment( api_key=litellm_params.get("api_key"), @@ -5032,23 +5023,31 @@ class BaseLLMHTTPHandler: litellm_params=litellm_params, ) try: - latest_page = client.get(url=url, headers=next_headers, params=params) + latest_page = client.get(url=url, headers=next_headers, params=params, timeout=timeout) except Exception as e: # noqa: BLE001 # _handle_error maps every failure kind, like the first page's fetch raise self._handle_error(e=e, provider_config=provider_config) - yield latest_page - async def _following_async_list_files_pages( + async def _files_per_async_listing_page( self, - page: httpx.Response, + first_page: httpx.Response, provider_config: BaseFilesConfig, + logging_obj: LiteLLMLoggingObj, litellm_params: dict, # mutable-ok: handed to the files contract, which types it as a dict headers: dict, # mutable-ok: handed to validate_environment, which types it as a dict client: AsyncHTTPHandler, - ) -> AsyncIterator[httpx.Response]: - latest_page = page # rebind-ok: advances one page per loop turn - while next_request := provider_config.transform_list_files_next_request( - raw_response=latest_page, optional_params={}, litellm_params=litellm_params - ): + timeout: float | httpx.Timeout | None, + ) -> AsyncIterator[list[OpenAIFileObject]]: # mutable-ok: each page arrives as the list the files contract returns + latest_page = first_page # rebind-ok: advances one page per loop turn + listed_count = 0 # rebind-ok: grows per page so the listing stops at MAX_FILE_LIST_LIMIT, OpenAI's ceiling + while True: + page_files = provider_config.transform_list_files_response( + raw_response=latest_page, logging_obj=logging_obj, litellm_params=litellm_params + ) + yield page_files[: MAX_FILE_LIST_LIMIT - listed_count] + listed_count += len(page_files) + next_request = self._next_listing_request(latest_page, provider_config, litellm_params, listed_count) + if next_request is None: + return url, params = next_request next_headers = provider_config.validate_environment( api_key=litellm_params.get("api_key"), @@ -5059,10 +5058,22 @@ class BaseLLMHTTPHandler: litellm_params=litellm_params, ) try: - latest_page = await client.get(url=url, headers=next_headers, params=params) + latest_page = await client.get(url=url, headers=next_headers, params=params, timeout=timeout) except Exception as e: # noqa: BLE001 # _handle_error maps every failure kind, like the first page's fetch raise self._handle_error(e=e, provider_config=provider_config) - yield latest_page + + def _next_listing_request( + self, + latest_page: httpx.Response, + provider_config: BaseFilesConfig, + litellm_params: dict, # mutable-ok: handed to the files contract, which types it as a dict + listed_count: int, + ) -> tuple[str, dict[str, str]] | None: # mutable-ok: the base files contract returns the query as a dict + if listed_count >= MAX_FILE_LIST_LIMIT: + return None + return provider_config.transform_list_files_next_request( + raw_response=latest_page, optional_params={}, litellm_params=litellm_params + ) def retrieve_file_content( self, diff --git a/litellm/proxy/openai_files_endpoints/common_utils.py b/litellm/proxy/openai_files_endpoints/common_utils.py index 992ed0d814d..4b4a99f849a 100644 --- a/litellm/proxy/openai_files_endpoints/common_utils.py +++ b/litellm/proxy/openai_files_endpoints/common_utils.py @@ -15,6 +15,7 @@ from typing import ( runtime_checkable, ) +from litellm.constants import MAX_FILE_LIST_LIMIT from litellm.proxy._types import ProxyException from litellm.repositories.table_repositories import ( ManagedFileRepository, @@ -33,8 +34,6 @@ if TYPE_CHECKING: from litellm.types.utils import LiteLLMBatch -MAX_FILE_LIST_LIMIT: Final = 10000 - FILE_LIST_CONTINUATION_CHUNK_SIZE: Final = 500 diff --git a/tests/test_litellm/llms/bedrock/files/test_bedrock_files_transformation.py b/tests/test_litellm/llms/bedrock/files/test_bedrock_files_transformation.py index 8e564280bf7..9bf9e468dca 100644 --- a/tests/test_litellm/llms/bedrock/files/test_bedrock_files_transformation.py +++ b/tests/test_litellm/llms/bedrock/files/test_bedrock_files_transformation.py @@ -3178,6 +3178,8 @@ class TestBedrockFileListTransformation: def _assert_paged_listing(self, first_page, last_page, files): assert (first_page.call_count, last_page.call_count) == (1, 1) + read_timeouts = [call.request.extensions["timeout"]["read"] for call in (*first_page.calls, *last_page.calls)] + assert read_timeouts == [12.0, 12.0] assert "continuation-token" not in str(first_page.calls[0].request.url) last_request = last_page.calls[0].request assert _sent_signature(last_request.headers) == _s3_signature_for( @@ -3198,6 +3200,7 @@ class TestBedrockFileListTransformation: files = litellm.file_list( custom_llm_provider="bedrock", purpose="batch", + timeout=12, **_trusted_bucket_snapshot(s3_bucket_name="my-bucket"), ) @@ -3219,7 +3222,86 @@ class TestBedrockFileListTransformation: files = await litellm.afile_list( custom_llm_provider="bedrock", purpose="batch", + timeout=12, **_trusted_bucket_snapshot(s3_bucket_name="my-bucket"), ) self._assert_paged_listing(first_page, last_page, files) + + OVERSIZED_PAGE_SIZE = 3000 + OVERSIZED_PAGE_COUNT = 6 + + def _oversized_listing_page(self, page_index: int) -> bytes: + contents = "".join( + f"litellm-bedrock-files/page-{page_index}/obj-{index}.jsonl" + "2026-09-01T10:00:00.000Z1" + for index in range(self.OVERSIZED_PAGE_SIZE) + ) + continuation = ( + f"truepage-{page_index + 1}" + if page_index < self.OVERSIZED_PAGE_COUNT - 1 + else "false" + ) + return ( + '' + '' + f"{continuation}{contents}" + ).encode() + + def _mock_oversized_listing(self, respx_module): + import httpx + + def page_for(request): + token = request.url.params.get("continuation-token", "page-0") + return httpx.Response(200, content=self._oversized_listing_page(int(token.removeprefix("page-")))) + + return respx_module.get(self.BUCKET_URL, params__contains=self.BATCH_QUERY).mock(side_effect=page_for) + + def _assert_capped_listing(self, route, files): + from litellm.constants import MAX_FILE_LIST_LIMIT + + pages_needed = -(-MAX_FILE_LIST_LIMIT // self.OVERSIZED_PAGE_SIZE) + last_index = MAX_FILE_LIST_LIMIT - (pages_needed - 1) * self.OVERSIZED_PAGE_SIZE - 1 + assert pages_needed < self.OVERSIZED_PAGE_COUNT + assert route.call_count == pages_needed + assert len(files) == MAX_FILE_LIST_LIMIT + assert files[-1].id == f"s3://my-bucket/litellm-bedrock-files/page-{pages_needed - 1}/obj-{last_index}.jsonl" + + def test_file_list_stops_at_the_openai_listing_ceiling(self, monkeypatch): + import respx + + import litellm + + monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False) + monkeypatch.delenv("AWS_S3_OUTPUT_BUCKET_NAME", raising=False) + + with respx.mock: + route = self._mock_oversized_listing(respx) + files = litellm.file_list( + custom_llm_provider="bedrock", + purpose="batch", + **_trusted_bucket_snapshot(s3_bucket_name="my-bucket"), + ) + + self._assert_capped_listing(route, files) + + @pytest.mark.asyncio + async def test_afile_list_stops_at_the_openai_listing_ceiling(self, monkeypatch): + import respx + + import litellm + + monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False) + monkeypatch.delenv("AWS_S3_OUTPUT_BUCKET_NAME", raising=False) + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + litellm.in_memory_llm_clients_cache.flush_cache() + + with respx.mock: + route = self._mock_oversized_listing(respx) + files = await litellm.afile_list( + custom_llm_provider="bedrock", + purpose="batch", + **_trusted_bucket_snapshot(s3_bucket_name="my-bucket"), + ) + + self._assert_capped_listing(route, files)