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

Merge b4919d9bd7 into 0c98afa780

Browse source
This commit is contained in:
Mateo Wang 2026-09-12 14:56:13 -04:00 • committed by GitHub
commit 3faed2d314
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 1864 additions and 69 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
enterprise/litellm_enterprise/proxy/hooks/managed_files.py
View file

@ -27,10 +27,13 @@ import litellm
from litellm import Router, verbose_logger
from litellm._uuid import uuid
from litellm.caching.caching import DualCache
from litellm.constants import MAX_FILE_LIST_LIMIT
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.prompt_templates.common_utils import (
extract_file_metadata,
)
from openai.types.file_deleted import FileDeleted
from litellm.llms.base_llm.files.transformation import BaseFileEndpoints
from litellm.llms.base_llm.managed_resources.isolation import (
build_list_page,
@ -48,7 +51,6 @@ from litellm.proxy._types import (
from litellm.proxy.openai_files_endpoints.common_utils import (
BATCH_CREATE_HIDDEN_PARAM,
FILE_LIST_CONTINUATION_CHUNK_SIZE,
MAX_FILE_LIST_LIMIT,
_is_base64_encoded_unified_file_id,
apply_unified_file_ids,
decode_model_from_file_id,
@ -1787,7 +1789,7 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
litellm_parent_otel_span: Optional[Span],
llm_router: Router,
**data: Dict,
) -> OpenAIFileObject:
) -> FileDeleted:
# Check if file deletion should be blocked due to batch references
await self._check_file_deletion_allowed(file_id)
@ -1795,7 +1797,6 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
# file_id = convert_b64_uid_to_unified_uid(file_id)
model_file_id_mapping = await self.get_model_file_id_mapping([file_id], litellm_parent_otel_span)
delete_response = None
specific_model_file_id_mapping = model_file_id_mapping.get(file_id)
if specific_model_file_id_mapping:
# Remove conflicting keys from data to avoid duplicate keyword arguments
@ -1810,23 +1811,14 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
else {}
),
}
delete_response = await llm_router.afile_delete(model=model_id, file_id=model_file_id, **delete_data)
await llm_router.afile_delete(model=model_id, file_id=model_file_id, **delete_data)
stored_file_object = await self.delete_unified_file_id(file_id, litellm_parent_otel_span)
await self.delete_unified_file_id(file_id, litellm_parent_otel_span)
# Record successful deletion metric only on actual success
if stored_file_object or delete_response:
prom_logger = self._get_prometheus_logger()
if prom_logger:
prom_logger.record_managed_file_deleted(result="success")
if stored_file_object:
return OpenAIFileObject.model_validate(stored_file_object).model_copy(update={"id": file_id})
elif delete_response:
delete_response.id = file_id
return delete_response
else:
raise Exception(f"LiteLLM Managed File object with id={file_id} not found")
prom_logger = self._get_prometheus_logger()
if prom_logger:
prom_logger.record_managed_file_deleted(result="success")
return FileDeleted(id=file_id, object="file", deleted=True)
async def afile_content(
self,

1
litellm/constants.py
View file

@ -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))
budget_reservation_disabled_info_emitted = False

4
litellm/files/main.py
View file

@ -682,6 +682,10 @@ def file_list(
)
if provider_config is not None:
litellm_params_dict: Final = get_litellm_params(**kwargs)
add_trusted_model_credentials_to_litellm_params(
litellm_params_dict=litellm_params_dict,
kwargs=kwargs,
)
litellm_params_dict["api_key"] = optional_params.api_key
litellm_params_dict["api_base"] = optional_params.api_base

13
litellm/llms/base_llm/files/transformation.py
View file

@ -1,5 +1,5 @@
from abc import ABC, abstractmethod
from collections.abc import Iterator
from collections.abc import Iterator, Mapping
from typing import TYPE_CHECKING, Any, Union
import httpx
@ -160,6 +160,15 @@ class BaseFilesConfig(BaseConfig):
) -> tuple[str, dict]:
"""Transform file list request into provider-specific format."""
def transform_list_files_next_request(
self,
raw_response: httpx.Response,
optional_params: Mapping[str, object],
litellm_params: dict, # mutable-ok: carries provider stashes from the request transform to the response one
) -> tuple[str, dict[str, str]] | None:
"""Request for the page after `raw_response`, or None once the listing is complete."""
return None
@abstractmethod
def transform_list_files_response(
self,
@ -258,7 +267,7 @@ class BaseFileEndpoints(ABC):
litellm_parent_otel_span: Span | None,
llm_router: Router,
**data: dict,
) -> OpenAIFileObject:
) -> FileDeleted:
pass
@abstractmethod

282
litellm/llms/bedrock/files/transformation.py
View file

@ -1,14 +1,18 @@
import base64
import json
import os
import posixpath
import time
import xml.etree.ElementTree as ET
from collections.abc import Iterable, Mapping, MutableMapping, Sequence
from contextlib import suppress
from dataclasses import dataclass
from datetime import datetime
from functools import cache
from itertools import chain
from types import MappingProxyType
from typing import Any, Final, Literal, TypeAlias, TypedDict
from urllib.parse import unquote
from urllib.parse import quote, unquote, urlencode
import httpx
from httpx import Headers, Response
@ -23,6 +27,7 @@ from litellm.files.utils import FilesAPIUtils
from litellm.litellm_core_utils.aws_partition import get_aws_dns_suffix
from litellm.litellm_core_utils.cloud_storage_security import (
BEDROCK_MANAGED_S3_BATCH_PREFIX,
BEDROCK_MANAGED_S3_OUTPUT_PREFIX,
BEDROCK_MANAGED_S3_PREFIXES,
BEDROCK_MANAGED_S3_UPLOAD_PREFIX,
build_managed_cloud_object_name,
@ -62,6 +67,10 @@ from ..common_utils import BedrockError, merge_bedrock_aws_request_params, resol
S3_SIGNED_REQUEST_HEADERS_PARAM: Final = "_s3_signed_request_headers"
LIST_FILES_PURPOSE_PARAM: Final = "_s3_list_files_purpose"
LIST_FILES_LOCATION_PARAM: Final = "_s3_list_files_location"
class _S3DeleteContext(BaseModel):
file_id: str = Field(min_length=1)
@ -152,6 +161,13 @@ class _BedrockS3RequestParams(BaseModel):
s3_endpoint_url: str | None = None
@dataclass(frozen=True, slots=True)
class _S3RequestTarget:
endpoint_url: str
aws_region_name: str
request_params: _BedrockS3RequestParams
class _TrustedS3ModelCredentials(BaseModel):
"""The S3 buckets the server trusts file ids against, from the deployment snapshot."""
@ -248,6 +264,128 @@ def _validate_file_id_against_configured_buckets(
return validate_against(configured_bucket_names[-1])
_REJECTED_FILE_ID_REQUEST_URL: Final = "https://litellm.ai"
def _rejected_file_id(reason: ValueError) -> BedrockError:
message: Final = str(reason)
return BedrockError(
status_code=400,
message=message,
response=httpx.Response(
status_code=400,
text=message,
request=httpx.Request(method="GET", url=_REJECTED_FILE_ID_REQUEST_URL),
),
)
def _resolve_managed_s3_object(file_id: str, litellm_params: Mapping[str, object]) -> tuple[str, str]:
configured_bucket_names: Final = get_configured_s3_bucket_names(litellm_params)
allow_legacy_cloud_file_ids: Final = should_allow_legacy_cloud_file_ids(litellm_params)
try:
return _validate_file_id_against_configured_buckets(
s3_uri=extract_s3_uri_from_file_id(file_id),
configured_bucket_names=configured_bucket_names,
allow_legacy_cloud_file_ids=allow_legacy_cloud_file_ids,
)
except ValueError as reason:
raise _rejected_file_id(reason) from reason
_ANY_MANAGED_LISTING_PREFIX: Final = os.path.commonprefix(BEDROCK_MANAGED_S3_PREFIXES)
_MANAGED_LISTING_PREFIX_BY_PURPOSE: Final = MappingProxyType(
{
"batch": os.path.commonprefix((BEDROCK_MANAGED_S3_BATCH_PREFIX, BEDROCK_MANAGED_S3_UPLOAD_PREFIX)),
"batch_output": BEDROCK_MANAGED_S3_OUTPUT_PREFIX,
}
)
_EMPTY_LISTING_QUERY: Final = (("list-type", "2"), ("max-keys", "0"))
def _managed_listing_prefix(configured_prefix: str, purpose: str | None) -> str | None:
managed_prefix: Final = _MANAGED_LISTING_PREFIX_BY_PURPOSE.get(purpose) if purpose else _ANY_MANAGED_LISTING_PREFIX
if managed_prefix is None:
return None
return f"{configured_prefix}/{managed_prefix}" if configured_prefix else managed_prefix
def _listing_query(configured_prefix: str, purpose: str | None) -> tuple[tuple[str, str], ...]:
listing_prefix: Final = _managed_listing_prefix(configured_prefix, purpose)
if listing_prefix is None:
return _EMPTY_LISTING_QUERY
return (("list-type", "2"), ("prefix", listing_prefix))
def _requested_listing_purpose(litellm_params: Mapping[str, object]) -> str | None:
requested_purpose: Final = litellm_params.get(LIST_FILES_PURPOSE_PARAM)
return requested_purpose if isinstance(requested_purpose, str) else None
def _walked_listing_purpose(litellm_params: Mapping[str, object]) -> str | None:
walked_purpose: Final = litellm_params.get(LIST_FILES_LOCATION_PARAM)
return walked_purpose if isinstance(walked_purpose, str) else _requested_listing_purpose(litellm_params)
def _output_location_still_unlisted(litellm_params: Mapping[str, object]) -> bool:
if _walked_listing_purpose(litellm_params) is not None:
return False
return _listing_bucket_name(litellm_params, "batch_output") != _listing_bucket_name(litellm_params, None)
def _listing_bucket_name(litellm_params: Mapping[str, object], purpose: str | None) -> str:
if purpose != "batch_output":
return get_configured_s3_bucket_name(litellm_params)
trusted: Final = _trusted_s3_model_credentials(litellm_params)
return (
trusted.s3_output_bucket_name
or os.getenv("AWS_S3_OUTPUT_BUCKET_NAME")
or get_configured_s3_bucket_name(litellm_params)
)
def _listed_object_created_at(entry: ET.Element) -> int:
last_modified: Final = entry.findtext("{*}LastModified")
if not last_modified:
return 0
return int(datetime.fromisoformat(last_modified.replace("Z", "+00:00")).timestamp())
def _listed_managed_file(
entry: ET.Element,
bucket_name: str,
configured_bucket_name: str,
allow_legacy_cloud_file_ids: bool,
) -> OpenAIFileObject | None:
object_key: Final = entry.findtext("{*}Key")
if not object_key:
return None
file_id: Final = f"s3://{bucket_name}/{object_key}"
try:
validate_managed_cloud_file_id(
file_id=file_id,
scheme="s3://",
configured_bucket_name=configured_bucket_name,
allowed_object_prefixes=BEDROCK_MANAGED_S3_PREFIXES,
allow_legacy_cloud_file_ids=allow_legacy_cloud_file_ids,
)
except ValueError:
return None
_, configured_prefix = split_configured_cloud_bucket_name(configured_bucket_name)
relative_key: Final = object_key[len(configured_prefix) + 1 :] if configured_prefix else object_key
return OpenAIFileObject(
id=file_id,
bytes=int(entry.findtext("{*}Size") or 0),
created_at=_listed_object_created_at(entry),
filename=posixpath.basename(object_key),
object="file",
purpose="batch_output" if relative_key.startswith(BEDROCK_MANAGED_S3_OUTPUT_PREFIX) else "batch",
status="uploaded",
)
def _uploaded_object_size(litellm_params: Mapping[str, object], raw_response: Response) -> int:
"""
S3 answers PutObject with an empty body, so the stored object size comes from the
@ -1213,18 +1351,86 @@ class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig):
def transform_list_files_request(
self,
purpose: str | None,
optional_params: dict,
litellm_params: dict,
) -> tuple[str, dict]:
raise NotImplementedError("BedrockFilesConfig does not support file listing")
optional_params: Mapping[str, object],
litellm_params: MutableMapping[str, object],
) -> tuple[str, dict[str, str]]:
litellm_params[LIST_FILES_PURPOSE_PARAM] = purpose # rebind-ok: handed to the response transform
litellm_params[LIST_FILES_LOCATION_PARAM] = purpose # rebind-ok: names the location the next page walks
return self._signed_listing_request(purpose, optional_params, litellm_params, continuation_token=None)
def transform_list_files_next_request(
self,
raw_response: httpx.Response,
optional_params: Mapping[str, object],
litellm_params: MutableMapping[str, object],
) -> tuple[str, dict[str, str]] | None:
if raw_response.status_code >= 400:
return None
continuation_token: Final = ET.fromstring(raw_response.content).findtext("{*}NextContinuationToken")
if continuation_token:
return self._signed_listing_request(
_walked_listing_purpose(litellm_params), optional_params, litellm_params, continuation_token
)
if not _output_location_still_unlisted(litellm_params):
return None
litellm_params[LIST_FILES_LOCATION_PARAM] = "batch_output" # rebind-ok: the input location is fully listed
return self._signed_listing_request("batch_output", optional_params, litellm_params, continuation_token=None)
def _signed_listing_request(
self,
purpose: str | None,
optional_params: Mapping[str, object],
litellm_params: MutableMapping[str, object],
continuation_token: str | None,
) -> tuple[str, dict[str, str]]:
bucket_name, configured_prefix = split_configured_cloud_bucket_name(
_listing_bucket_name(litellm_params, purpose)
)
target: Final = self._s3_request_target(optional_params=optional_params, litellm_params=litellm_params)
url: Final = f"{target.endpoint_url}/{bucket_name}/"
listing_query: Final = _listing_query(configured_prefix, purpose)
continuation_query: Final = (("continuation-token", continuation_token),) if continuation_token else ()
query: Final[dict[str, str]] = dict( # mutable-ok: the base files contract returns the query as a dict
listing_query + continuation_query
)
signed_headers: Final = self._sign_s3_request_without_body(
method="GET",
api_base=f"{url}?{urlencode(query, quote_via=quote, safe='')}",
aws_region_name=target.aws_region_name,
request_params=target.request_params,
)
litellm_params[S3_SIGNED_REQUEST_HEADERS_PARAM] = signed_headers # rebind-ok: handed to validate_environment
return url, query
def transform_list_files_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
litellm_params: dict,
litellm_params: Mapping[str, object],
) -> list[OpenAIFileObject]:
raise NotImplementedError("BedrockFilesConfig does not support file listing")
if raw_response.status_code >= 400:
raise BedrockError(
status_code=raw_response.status_code,
message=raw_response.text,
headers=raw_response.headers,
response=raw_response,
)
purpose: Final = _requested_listing_purpose(litellm_params)
configured_bucket_name: Final = _listing_bucket_name(litellm_params, _walked_listing_purpose(litellm_params))
allow_legacy_cloud_file_ids: Final = should_allow_legacy_cloud_file_ids(litellm_params)
listing: Final = ET.fromstring(raw_response.content)
bucket_name: Final = (
listing.findtext("{*}Name") or split_configured_cloud_bucket_name(configured_bucket_name)[0]
)
listed_files: Final = (
_listed_managed_file(entry, bucket_name, configured_bucket_name, allow_legacy_cloud_file_ids)
for entry in listing.iterfind("{*}Contents")
)
return [ # mutable-ok: the base files contract returns a list
listed_file
for listed_file in listed_files
if listed_file is not None and (purpose is None or listed_file.purpose == purpose)
]
def transform_file_content_request(
self,
@ -1255,39 +1461,54 @@ class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig):
optional_params: Mapping[str, object],
litellm_params: MutableMapping[str, object],
) -> tuple[str, dict[str, str]]:
s3_uri: Final = extract_s3_uri_from_file_id(file_id)
bucket_name, object_key = _validate_file_id_against_configured_buckets(
s3_uri=s3_uri,
configured_bucket_names=get_configured_s3_bucket_names(litellm_params),
allow_legacy_cloud_file_ids=should_allow_legacy_cloud_file_ids(litellm_params),
bucket_name, object_key = _resolve_managed_s3_object(file_id=file_id, litellm_params=litellm_params)
target: Final = self._s3_request_target(optional_params=optional_params, litellm_params=litellm_params)
url: Final = f"{target.endpoint_url}/{bucket_name}/{encode_s3_object_key_for_url(object_key)}"
signed_headers: Final = self._sign_s3_request_without_body(
method=method,
api_base=url,
aws_region_name=target.aws_region_name,
request_params=target.request_params,
)
litellm_params[S3_SIGNED_REQUEST_HEADERS_PARAM] = signed_headers # rebind-ok: handed to validate_environment
return url, {} # mutable-ok: the base files contract returns the query as a dict
request_params: Final = _BedrockS3RequestParams.model_validate({**litellm_params, **optional_params})
def _s3_request_target(
self,
optional_params: Mapping[str, object],
litellm_params: Mapping[str, object],
) -> _S3RequestTarget:
"""
The shared files handler passes optional_params={}, so AWS credentials and
region arrive via litellm_params here (unlike the upload path).
s3_region_name wins over aws_region_name, same priority as get_complete_file_url.
"""
request_params: Final = _BedrockS3RequestParams.model_validate(
MappingProxyType({**litellm_params, **optional_params})
)
region_preference: Final = request_params.s3_region_name or request_params.aws_region_name
region_params: Final[dict[str, str | None]] = {"aws_region_name": region_preference}
aws_region_name: Final = self._get_aws_region_name(optional_params=region_params, model="")
s3_endpoint_url: Final = (
aws_region_name: Final = self._get_aws_region_name(
optional_params={"aws_region_name": region_preference}, # mutable-ok: BaseAWSLLM takes a dict
model="",
)
endpoint_url: Final = (
request_params.s3_endpoint_url or f"https://s3.{aws_region_name}.{get_aws_dns_suffix(aws_region_name)}"
).rstrip("/")
url: Final = f"{s3_endpoint_url}/{bucket_name}/{encode_s3_object_key_for_url(object_key)}"
litellm_params[S3_SIGNED_REQUEST_HEADERS_PARAM] = self._sign_s3_request_without_body(
api_base=url,
aws_region_name=aws_region_name,
request_params=request_params,
method=method,
return _S3RequestTarget(
endpoint_url=endpoint_url, aws_region_name=aws_region_name, request_params=request_params
)
return url, {}
def _sign_s3_request_without_body(
self,
method: Literal["GET", "DELETE"],
api_base: str,
aws_region_name: str,
request_params: _BedrockS3RequestParams,
method: Literal["GET", "DELETE"] = "GET",
) -> dict[str, str]:
) -> Mapping[str, str]:
"""
SigV4-sign a bodiless S3 request (GetObject, DeleteObject, ListObjectsV2),
mirroring `_sign_s3_request` (PUT).
"""
try:
import hashlib
@ -1313,11 +1534,11 @@ class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig):
aws_request: Final = AWSRequest( # any-ok: botocore AWSRequest is untyped
method=method,
url=api_base,
headers={"x-amz-content-sha256": empty_body_hash},
headers={"x-amz-content-sha256": empty_body_hash}, # mutable-ok: botocore AWSRequest takes a dict
)
auth: Final = S3SigV4Auth(credentials, "s3", aws_region_name) # any-ok: botocore untyped
auth.add_auth(aws_request) # any-ok: botocore request mutation is untyped
return dict(aws_request.headers) # any-ok: botocore headers are untyped
return MappingProxyType(dict(aws_request.headers)) # any-ok: botocore headers are untyped
def transform_file_content_response(
self,
@ -1330,6 +1551,7 @@ class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig):
status_code=raw_response.status_code,
message=raw_response.text,
headers=raw_response.headers,
response=raw_response,
)
return HttpxBinaryResponseContent(response=raw_response)

106
litellm/llms/custom_httpx/llm_http_handler.py
View file

@ -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,
@ -4981,15 +4981,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)
return provider_config.transform_list_files_response(
raw_response=response,
logging_obj=logging_obj,
litellm_params=litellm_params,
files_per_page: Final = self._files_per_listing_page(
response, provider_config, logging_obj, litellm_params, headers, sync_httpx_client, timeout
)
return [ # mutable-ok: the files contract returns the listing as a list
listed_file for page_files in files_per_page for listed_file in page_files
]
async def async_list_files(
self,
@ -5037,16 +5038,101 @@ 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)
return provider_config.transform_list_files_response(
raw_response=response,
logging_obj=logging_obj,
files_per_page: Final = self._files_per_async_listing_page(
response, provider_config, logging_obj, litellm_params, headers, async_httpx_client, timeout
)
return [ # mutable-ok: the files contract returns the listing as a list
listed_file async for page_files in files_per_page for listed_file in page_files
]
def _files_per_listing_page(
self,
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: HTTPHandler,
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 = self._next_listing_page_headers(provider_config, headers, litellm_params)
try:
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)
async def _files_per_async_listing_page(
self,
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,
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 = self._next_listing_page_headers(provider_config, headers, litellm_params)
try:
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)
def _next_listing_page_headers(
self,
provider_config: BaseFilesConfig,
headers: dict, # mutable-ok: handed to validate_environment, which types it as a dict
litellm_params: dict, # mutable-ok: handed to the files contract, which types it as a dict
) -> dict: # mutable-ok: validate_environment returns the header dict the files contract types
return provider_config.validate_environment(
api_key=litellm_params.get("api_key"),
headers=headers,
model="",
messages=[],
optional_params={},
litellm_params=litellm_params,
)
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,
file_content_request: "FileContentRequest",

3
litellm/proxy/openai_files_endpoints/common_utils.py
View file

@ -16,6 +16,7 @@ from typing import (
)
from litellm.batches.batch_utils import batch_cost_is_final
from litellm.constants import MAX_FILE_LIST_LIMIT
from litellm.proxy._types import ProxyException
from litellm.repositories.table_repositories import (
ManagedFileRepository,
@ -34,8 +35,6 @@ if TYPE_CHECKING:
from litellm.types.utils import LiteLLMBatch
MAX_FILE_LIST_LIMIT: Final = 10000
FILE_LIST_CONTINUATION_CHUNK_SIZE: Final = 500
BATCH_CREATE_HIDDEN_PARAM: Final = "batch_create"

19
litellm/proxy/openai_files_endpoints/files_endpoints.py
View file

@ -33,6 +33,7 @@ from litellm.litellm_core_utils.cloud_storage_security import (
)
from litellm.litellm_core_utils.core_helpers import get_or_create_metadata_bucket
from litellm.llms.base_llm.files.transformation import BaseFileEndpoints
from litellm.llms.base_llm.managed_resources.isolation import build_list_page
from litellm.proxy._types import *
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
@ -90,6 +91,7 @@ from litellm.router import Router
from litellm.types.llms.openai import (
CREATE_FILE_REQUESTS_PURPOSE,
FileExpiresAfter,
FileListPage,
OpenAIFileObject,
OpenAIFilesPurpose,
)
@ -97,6 +99,7 @@ from litellm.types.llms.openai import (
router: Final = APIRouter()
_MAX_BATCH_FILE_SIZE_MB_ADAPTER: Final = TypeAdapter(int | None)
_LISTED_FILES_ADAPTER: Final = TypeAdapter(list[OpenAIFileObject])
class UploadedFileInfo(TypedDict):
@ -1287,6 +1290,11 @@ async def delete_file(
user_api_key_dict=user_api_key_dict,
managed_files_obj=proxy_logging_obj.get_proxy_hook("managed_files"),
)
if is_managed_cloud_storage_uri(file_id) and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN:
raise HTTPException(
status_code=403,
detail="Raw cloud storage file ids can only be deleted by a proxy admin key. Use the LiteLLM managed file id returned when the file was created.",
)
custom_llm_provider: Final = (
provider
@ -1446,6 +1454,12 @@ async def delete_file(
)
def _as_file_list_page(response: object) -> object:
if not isinstance(response, list):
return response
return FileListPage(**build_list_page(_LISTED_FILES_ADAPTER.validate_python(response)))
@router.get(
"/{provider}/v1/files",
dependencies=[Depends(user_api_key_auth)],
@ -1524,7 +1538,7 @@ async def list_files(
if should_route and credentials is not None:
# Use model-based routing with credentials from config
prepare_data_with_credentials(data=data, credentials=credentials)
prepare_data_with_credentials(data=data, credentials=credentials, include_internal_credentials=True)
response = await litellm.afile_list(
custom_llm_provider=credentials["custom_llm_provider"],
purpose=purpose,
@ -1550,7 +1564,7 @@ async def list_files(
model_id=target_model_names_list[0],
operation_context="file list",
)
prepare_data_with_credentials(data=data, credentials=credentials)
prepare_data_with_credentials(data=data, credentials=credentials, include_internal_credentials=True)
response = await litellm.afile_list(
custom_llm_provider=credentials["custom_llm_provider"],
purpose=purpose,
@ -1592,6 +1606,7 @@ async def list_files(
status_code=500,
detail="Either 'provider' or 'target_model_names' must be provided e.g. `?target_model_names=gpt-4o`",
)
response = _as_file_list_page(response) # rebind-ok: each dispatch branch above binds response
## POST CALL HOOKS ###
_response: Final = await proxy_logging_obj.post_call_success_hook(

99
tests/test_litellm/enterprise/proxy/test_managed_files_hook.py
View file

@ -1193,7 +1193,7 @@ async def test_afile_delete_returns_managed_id_for_stored_provider_output():
assert response.id == unified_file_id
assert response.object == "file"
assert response.filename == stored_file.filename
assert response.deleted is True
assert stored_file.id == provider_file_id
router.afile_delete.assert_awaited_once_with(model="model-123", file_id=provider_file_id)
table.delete.assert_awaited_once_with(where={"unified_file_id": unified_file_id})
@ -1730,3 +1730,100 @@ async def test_batch_retrieve_hook_does_not_claim_attribution():
managed_files.store_unified_object_id.assert_awaited_once()
assert managed_files.store_unified_object_id.await_args.kwargs["persist_attribution"] is False
@pytest.mark.asyncio
async def test_afile_delete_passes_trusted_model_credentials_to_router():
"""
afile_delete must hand the deployment's credential snapshot to the router
call, since Bedrock validates the s3:// file id against the bucket in it.
"""
from types import MappingProxyType
managed_files = _make_managed_files_instance()
unified_file_id = "unified-file-id"
s3_uri = "s3://my-bucket/litellm-bedrock-files/job-123/input.jsonl"
managed_files.get_model_file_id_mapping = AsyncMock(return_value={unified_file_id: {"model-123": s3_uri}})
managed_files.delete_unified_file_id = AsyncMock(return_value=_make_file_object(unified_file_id))
mock_router = MagicMock()
mock_router.get_deployment_credentials_with_provider = MagicMock(
return_value={
"custom_llm_provider": "bedrock",
"s3_bucket_name": "my-bucket",
"aws_region_name": "us-west-2",
}
)
mock_router.afile_delete = AsyncMock(return_value=MagicMock())
await managed_files.afile_delete(
file_id=unified_file_id,
litellm_parent_otel_span=None,
llm_router=mock_router,
)
call_kwargs = mock_router.afile_delete.call_args.kwargs
assert call_kwargs["model"] == "model-123"
assert call_kwargs["file_id"] == s3_uri
trusted_credentials = call_kwargs["_litellm_internal_model_credentials"]
assert isinstance(trusted_credentials, MappingProxyType)
assert trusted_credentials["s3_bucket_name"] == "my-bucket"
@pytest.mark.asyncio
async def test_afile_delete_bedrock_unified_id_end_to_end(monkeypatch):
"""
Proxy repro for deleting a Bedrock batch input file by unified id: the
s3:// object must be removed via a SigV4-signed S3 DELETE using the
deployment's s3_bucket_name (no AWS_S3_BUCKET_NAME env).
Regression test for "BedrockFilesConfig does not support file deletion"
raised on this path.
"""
import httpx
import respx
import litellm
from litellm import Router
monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
litellm.in_memory_llm_clients_cache.flush_cache()
router = Router(
model_list=[
{
"model_name": "bedrock-claude",
"litellm_params": {
"model": "bedrock/anthropic.claude-3-5-sonnet-20240620-v1:0",
"aws_access_key_id": "AKIAEXAMPLE",
"aws_secret_access_key": "secret",
"aws_region_name": "us-west-2",
"s3_bucket_name": "my-bucket",
},
"model_info": {"id": "model-123"},
}
]
)
managed_files = _make_managed_files_instance()
unified_file_id = "unified-file-id"
s3_uri = "s3://my-bucket/litellm-bedrock-files/job-123/input.jsonl"
managed_files.get_model_file_id_mapping = AsyncMock(return_value={unified_file_id: {"model-123": s3_uri}})
managed_files.delete_unified_file_id = AsyncMock(return_value=_make_file_object(unified_file_id))
expected_url = "https://s3.us-west-2.amazonaws.com/my-bucket/litellm-bedrock-files/job-123/input.jsonl"
with respx.mock:
route = respx.delete(expected_url).mock(return_value=httpx.Response(204))
response = await managed_files.afile_delete(
file_id=unified_file_id,
litellm_parent_otel_span=None,
llm_router=router,
)
assert route.called
assert route.calls[0].request.headers["Authorization"].startswith("AWS4-HMAC-SHA256")
assert response.id == unified_file_id
assert response.model_dump() == {"id": unified_file_id, "object": "file", "deleted": True}
managed_files.delete_unified_file_id.assert_awaited_once_with(unified_file_id, None)

1078
tests/test_litellm/llms/bedrock/files/test_bedrock_files_transformation.py
View file

File diff suppressed because it is too large Load diff

300
tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py
View file

@ -2441,6 +2441,63 @@ def test_list_files_resolves_wildcard_deployment_credentials(
proxy_logging_obj.post_call_failure_hook.assert_not_called()
def test_list_files_by_model_returns_an_openai_page_for_a_provider_listing(
mocker: MockerFixture, monkeypatch, llm_router: Router
):
import litellm.proxy.proxy_server as ps
from litellm.proxy._types import LitellmUserRoles
from litellm.types.llms.openai import FileListPage
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)
monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", llm_router)
proxy_logging_obj.update_request_status = mocker.AsyncMock()
proxy_logging_obj.post_call_success_hook = mocker.AsyncMock(return_value=None)
proxy_logging_obj.post_call_failure_hook = mocker.AsyncMock()
listed_files = [
OpenAIFileObject(
id=f"file-{index}",
bytes=index,
created_at=index,
filename=f"{index}.jsonl",
object="file",
purpose="batch",
status="uploaded",
)
for index in (1, 2)
]
async def _mock_afile_list(**kwargs):
return list(listed_files)
monkeypatch.setattr(litellm, "afile_list", _mock_afile_list)
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?target_model_names=gpt-3.5-turbo",
headers={"Authorization": "Bearer test-key"},
)
finally:
app.dependency_overrides.pop(ps.user_api_key_auth, None)
assert response.status_code == 200, response.text
body = response.json()
assert body["object"] == "list"
assert [listed["id"] for listed in body["data"]] == ["file-1", "file-2"]
assert (body["first_id"], body["last_id"], body["has_more"]) == ("file-1", "file-2", False)
hook_response = proxy_logging_obj.post_call_success_hook.call_args.kwargs["response"]
assert isinstance(hook_response, FileListPage)
assert [listed.id for listed in hook_response.data] == ["file-1", "file-2"]
def test_list_files_model_routing_does_not_forward_custom_llm_provider_twice(
mocker: MockerFixture, monkeypatch, llm_router: Router
):
@ -4668,6 +4725,249 @@ def test_create_file_path_traversal_filename_rejected_before_forwarding(monkeypa
assert forwarded_calls == []
def test_list_files_target_model_names_passes_trusted_bedrock_credentials(
mocker: MockerFixture, monkeypatch
):
"""
GET /v1/files?target_model_names=<bedrock model> must hand the deployment's
immutable credential snapshot to litellm.afile_list, since Bedrock resolves
the S3 bucket to list from that snapshot rather than from request params.
"""
from types import MappingProxyType
import litellm.proxy.proxy_server as ps
from litellm.proxy._types import LitellmUserRoles
bedrock_router = Router(
model_list=[
{
"model_name": "bedrock-claude",
"litellm_params": {
"model": "bedrock/anthropic.claude-3-5-sonnet-20240620-v1:0",
"aws_access_key_id": "AKIAEXAMPLE",
"aws_secret_access_key": "secret",
"aws_region_name": "us-west-2",
"s3_bucket_name": "my-bucket",
},
},
]
)
proxy_logging_obj = setup_proxy_logging_object(monkeypatch, bedrock_router)
monkeypatch.setattr("litellm.proxy.proxy_server.master_key", None)
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", None)
monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", bedrock_router)
proxy_logging_obj.update_request_status = mocker.AsyncMock()
proxy_logging_obj.post_call_success_hook = mocker.AsyncMock(return_value=[])
proxy_logging_obj.post_call_failure_hook = mocker.AsyncMock()
captured_kwargs: dict = {}
async def _mock_afile_list(**kwargs):
captured_kwargs.update(kwargs)
return []
monkeypatch.setattr(litellm, "afile_list", _mock_afile_list)
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?target_model_names=bedrock-claude&purpose=batch",
headers={"Authorization": "Bearer test-key"},
)
finally:
app.dependency_overrides.pop(ps.user_api_key_auth, None)
assert response.status_code == 200, response.text
assert captured_kwargs["custom_llm_provider"] == "bedrock"
assert captured_kwargs["purpose"] == "batch"
trusted_credentials = captured_kwargs["_litellm_internal_model_credentials"]
assert isinstance(trusted_credentials, MappingProxyType)
assert trusted_credentials["s3_bucket_name"] == "my-bucket"
proxy_logging_obj.post_call_failure_hook.assert_not_called()
def test_delete_file_answers_400_for_an_id_outside_the_configured_bucket(mocker: MockerFixture, monkeypatch):
from urllib.parse import quote
import litellm.proxy.proxy_server as ps
from litellm.proxy._types import LitellmUserRoles
bedrock_router = Router(
model_list=[
{
"model_name": "bedrock-claude",
"litellm_params": {
"model": "bedrock/anthropic.claude-3-5-sonnet-20240620-v1:0",
"aws_access_key_id": "AKIAEXAMPLE",
"aws_secret_access_key": "secret",
"aws_region_name": "us-west-2",
"s3_bucket_name": "my-bucket",
},
},
]
)
proxy_logging_obj = setup_proxy_logging_object(monkeypatch, bedrock_router)
monkeypatch.setattr("litellm.proxy.proxy_server.master_key", None)
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", None)
monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", bedrock_router)
proxy_logging_obj.update_request_status = mocker.AsyncMock()
proxy_logging_obj.post_call_success_hook = mocker.AsyncMock(return_value=[])
proxy_logging_obj.post_call_failure_hook = mocker.AsyncMock()
app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth(
api_key="test-key",
user_role=LitellmUserRoles.PROXY_ADMIN,
user_id="test-user",
)
foreign_file_id: Final = quote("s3://other-bucket/litellm-bedrock-files/job-123/input.jsonl", safe="")
try:
with respx.mock:
response = client.delete(
f"/v1/files/{foreign_file_id}?model=bedrock-claude",
headers={"Authorization": "Bearer test-key"},
)
finally:
app.dependency_overrides.pop(ps.user_api_key_auth, None)
assert response.status_code == 400, response.text
assert "configured storage bucket" in response.json()["error"]["message"]
def _cloud_files_router() -> Router:
return Router(
model_list=[
{
"model_name": "bedrock-claude",
"litellm_params": {
"model": "bedrock/anthropic.claude-3-5-sonnet-20240620-v1:0",
"aws_access_key_id": "AKIAEXAMPLE",
"aws_secret_access_key": "secret",
"aws_region_name": "us-west-2",
"s3_bucket_name": "my-bucket",
},
},
{
"model_name": "vertex-gemini",
"litellm_params": {
"model": "vertex_ai/gemini-3.8-flash",
"vertex_project": "my-project",
"vertex_location": "us-central1",
"gcs_bucket_name": "my-gcs-bucket",
},
},
]
)
RAW_S3_FILE_ID: Final = "s3://my-bucket/litellm-batch-outputs/job-123/abc/input.jsonl.out"
RAW_GCS_FILE_ID: Final = "gs://my-gcs-bucket/litellm-vertex-files/publishers/google/models/gemini-3.8-flash/abc123"
@pytest.mark.parametrize(
("route_prefix", "raw_file_id", "model_name"),
(
("/bedrock/v1/files", RAW_S3_FILE_ID, "bedrock-claude"),
("/v1/files", RAW_S3_FILE_ID, "bedrock-claude"),
("/files", RAW_S3_FILE_ID, "bedrock-claude"),
("/vertex_ai/v1/files", RAW_GCS_FILE_ID, "vertex-gemini"),
),
)
def test_delete_file_answers_403_for_a_raw_cloud_id_from_a_non_admin_key(
mocker: MockerFixture, monkeypatch, route_prefix: str, raw_file_id: str, model_name: str
):
from urllib.parse import quote
import litellm.proxy.proxy_server as ps
from litellm.proxy._types import LitellmUserRoles
bedrock_router = _cloud_files_router()
proxy_logging_obj = setup_proxy_logging_object(monkeypatch, bedrock_router)
monkeypatch.setattr("litellm.proxy.proxy_server.master_key", None)
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", None)
monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", bedrock_router)
proxy_logging_obj.update_request_status = mocker.AsyncMock()
proxy_logging_obj.post_call_failure_hook = mocker.AsyncMock()
afile_delete = mocker.AsyncMock()
monkeypatch.setattr(litellm, "afile_delete", afile_delete)
app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth(
api_key="test-key",
user_role=LitellmUserRoles.INTERNAL_USER,
user_id="test-user",
models=["bedrock-claude", "vertex-gemini"],
)
try:
response = client.delete(
f"{route_prefix}/{quote(raw_file_id, safe='')}?model={model_name}",
headers={"Authorization": "Bearer test-key"},
)
finally:
app.dependency_overrides.pop(ps.user_api_key_auth, None)
assert response.status_code == 403, response.text
assert "proxy admin" in response.json()["error"]["message"]
afile_delete.assert_not_called()
def test_delete_file_forwards_a_raw_cloud_id_from_a_proxy_admin_key(mocker: MockerFixture, monkeypatch):
from urllib.parse import quote
import litellm.proxy.proxy_server as ps
from litellm.proxy._types import LitellmUserRoles
bedrock_router = _cloud_files_router()
proxy_logging_obj = setup_proxy_logging_object(monkeypatch, bedrock_router)
monkeypatch.setattr("litellm.proxy.proxy_server.master_key", None)
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", None)
monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", bedrock_router)
proxy_logging_obj.update_request_status = mocker.AsyncMock()
proxy_logging_obj.post_call_failure_hook = mocker.AsyncMock()
captured_kwargs: dict = {}
async def _mock_afile_delete(**kwargs):
captured_kwargs.update(kwargs)
return OpenAIFileObject(
id=RAW_S3_FILE_ID,
object="file",
bytes=2,
created_at=1234567890,
filename="input.jsonl.out",
purpose="batch_output",
status="processed",
)
monkeypatch.setattr(litellm, "afile_delete", _mock_afile_delete)
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.delete(
f"/bedrock/v1/files/{quote(RAW_S3_FILE_ID, safe='')}?model=bedrock-claude",
headers={"Authorization": "Bearer test-key"},
)
finally:
app.dependency_overrides.pop(ps.user_api_key_auth, None)
assert response.status_code == 200, response.text
assert captured_kwargs.get("file_id") == RAW_S3_FILE_ID
assert captured_kwargs.get("custom_llm_provider") == "bedrock"
proxy_logging_obj.post_call_failure_hook.assert_not_called()
def _setup_managed_file_route_answering_404(
mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch, llm_router: Router
) -> None: