# What is this? ## This hook is used to check for LiteLLM managed files in the request body, and replace them with model-specific file id import base64 import json from collections.abc import Mapping, Sequence from types import MappingProxyType from typing import ( TYPE_CHECKING, Any, Dict, Final, List, Literal, Optional, Protocol, TypedDict, Union, cast, ) from uuid import NAMESPACE_URL, uuid5 from fastapi import HTTPException from pydantic import ValidationError import litellm from litellm import Router, verbose_logger from litellm._uuid import uuid from litellm.caching.caching import DualCache from litellm.integrations.custom_logger import CustomLogger from litellm.litellm_core_utils.prompt_templates.common_utils import ( extract_file_metadata, ) from litellm.llms.base_llm.files.transformation import BaseFileEndpoints from litellm.llms.base_llm.managed_resources.isolation import ( build_list_page, build_owner_filter, can_access_resource, ) from litellm.proxy._types import ( CallTypes, LiteLLM_ManagedFileTable, LiteLLM_ManagedObjectTable, ProxyException, UserAPIKeyAuth, ) from litellm.proxy.openai_files_endpoints.common_utils import ( FILE_LIST_CONTINUATION_CHUNK_SIZE, MAX_FILE_LIST_LIMIT, _is_base64_encoded_unified_file_id, apply_unified_file_ids, ensure_batch_response_managed_file_ids, get_batch_id_from_unified_batch_id, get_content_type_from_file_object, get_model_id_from_unified_batch_id, map_raw_file_ids_to_unified, normalize_mime_type_for_provider, resolve_managed_output_file_model_name, validate_file_list_limit, validate_file_list_purpose, ) from litellm.proxy.pass_through_endpoints.llm_provider_handlers.batch_attribution import ( request_tags_from_metadata, ) from litellm.types.llms.openai import ( # pyright: ignore[reportAttributeAccessIssue] AllMessageValues, AsyncCursorPage, ChatCompletionFileObject, CreateFileRequest, FileListPage, FileObject, OpenAIFileObject, ResponsesAPIResponse, ) from litellm.types.utils import ( CallTypesLiteral, LiteLLMBatch, LiteLLMFineTuningJob, LLMResponseTypes, SpecialEnums, ) if TYPE_CHECKING: from litellm.types.llms.openai import HttpxBinaryResponseContent if TYPE_CHECKING: from opentelemetry.trace import Span as _Span from prisma.models import ( LiteLLM_ManagedObjectTable as PrismaManagedObjectRow, ) from litellm.proxy.utils import InternalUsageCache as _InternalUsageCache from litellm.proxy.utils import PrismaClient as _PrismaClient Span = Union[_Span, Any] InternalUsageCache = _InternalUsageCache PrismaClient = _PrismaClient else: Span = Any InternalUsageCache = Any PrismaClient = Any def _sanitized_parse_error(e: Exception) -> str: return ( str(e.errors(include_input=False, include_url=False, include_context=False)) if isinstance(e, ValidationError) else type(e).__name__ ) def _decode_json_blob(blob: object) -> object: return json.loads(blob) if isinstance(blob, str) else blob def _parse_managed_batch_row(row: "PrismaManagedObjectRow") -> Optional[LiteLLMBatch]: try: batch_obj: Final = LiteLLMBatch.model_validate(_decode_json_blob(row.file_object)) except Exception as e: verbose_logger.warning(f"Failed to parse batch object {row.unified_object_id}: {_sanitized_parse_error(e)}") return None batch_obj.id = row.unified_object_id return batch_obj def _parse_managed_file_object(raw_file_object: object, unified_file_id: str) -> Optional[OpenAIFileObject]: if raw_file_object is None: return None try: return OpenAIFileObject.model_validate(raw_file_object) except Exception as e: verbose_logger.warning(f"Failed to parse managed file object {unified_file_id}: {_sanitized_parse_error(e)}") return None class _ManagedFileRow(Protocol): unified_file_id: str file_object: OpenAIFileObject storage_backend: Optional[str] storage_url: Optional[str] created_by: Optional[str] team_id: Optional[str] def model_dump(self) -> Mapping[str, object]: ... class _ManagedFileTableActions(Protocol): async def find_first(self, where: Mapping[str, object]) -> Optional[_ManagedFileRow]: ... async def find_many( self, where: Mapping[str, object], take: int = ..., order: Union[Mapping[str, str], Sequence[Mapping[str, str]]] = ..., cursor: Mapping[str, str] = ..., skip: int = ..., ) -> Sequence[_ManagedFileRow]: ... async def upsert(self, where: Mapping[str, str], data: Mapping[str, Mapping[str, object]]) -> _ManagedFileRow: ... async def delete(self, where: Mapping[str, str]) -> Optional[_ManagedFileRow]: ... class _ManagedObjectTableActions(Protocol): async def find_first(self, where: Mapping[str, object]) -> "Optional[PrismaManagedObjectRow]": ... async def find_many( self, where: Mapping[str, object], take: int, order: Union[Mapping[str, str], Sequence[Mapping[str, str]]], cursor: Mapping[str, str] = ..., skip: int = ..., ) -> "Sequence[PrismaManagedObjectRow]": ... async def upsert( self, where: Mapping[str, str], data: Mapping[str, Mapping[str, object]] ) -> "PrismaManagedObjectRow": ... async def update_many(self, where: Mapping[str, object], data: Mapping[str, object]) -> int: ... class _CursorPageArgs(TypedDict, total=False): cursor: Mapping[str, str] skip: int def _managed_file_table(prisma_client: PrismaClient) -> _ManagedFileTableActions: return prisma_client.db.litellm_managedfiletable def _managed_object_table(prisma_client: PrismaClient) -> _ManagedObjectTableActions: return prisma_client.db.litellm_managedobjecttable class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints): # Class variables or attributes def __init__(self, internal_usage_cache: InternalUsageCache, prisma_client: PrismaClient): self.internal_usage_cache = internal_usage_cache self.prisma_client = prisma_client @staticmethod def _get_prometheus_logger(): """Find PrometheusLogger from litellm.callbacks, if registered.""" from litellm.integrations.prometheus import PrometheusLogger return PrometheusLogger.get_instance() async def store_unified_file_id( self, file_id: str, file_object: Optional[OpenAIFileObject], litellm_parent_otel_span: Optional[Span], model_mappings: Dict[str, str], user_api_key_dict: UserAPIKeyAuth, ) -> None: verbose_logger.info(f"Storing LiteLLM Managed File object with id={file_id} in cache") if file_object is not None: litellm_managed_file_object = LiteLLM_ManagedFileTable( unified_file_id=file_id, file_object=file_object, model_mappings=model_mappings, flat_model_file_ids=list(model_mappings.values()), created_by=user_api_key_dict.user_id, team_id=user_api_key_dict.team_id, updated_by=user_api_key_dict.user_id, ) await self.internal_usage_cache.async_set_cache( key=file_id, value=litellm_managed_file_object.model_dump(), litellm_parent_otel_span=litellm_parent_otel_span, ) ## STORE MODEL MAPPINGS IN DB db_data = { "unified_file_id": file_id, "model_mappings": json.dumps(model_mappings), "flat_model_file_ids": list(model_mappings.values()), "created_by": user_api_key_dict.user_id, "team_id": user_api_key_dict.team_id, "updated_by": user_api_key_dict.user_id, } update_data = { "model_mappings": json.dumps(model_mappings), "flat_model_file_ids": list(model_mappings.values()), "updated_by": user_api_key_dict.user_id, } if file_object is not None: file_object_json = file_object.model_dump_json() db_data["file_object"] = file_object_json update_data["file_object"] = file_object_json # Extract storage metadata from hidden params if present hidden_params = getattr(file_object, "_hidden_params", {}) or {} if "storage_backend" in hidden_params: db_data["storage_backend"] = hidden_params["storage_backend"] update_data["storage_backend"] = hidden_params["storage_backend"] if "storage_url" in hidden_params: db_data["storage_url"] = hidden_params["storage_url"] update_data["storage_url"] = hidden_params["storage_url"] verbose_logger.debug( f"Storage metadata: storage_backend={db_data.get('storage_backend')}, " f"storage_url={db_data.get('storage_url')}" ) result = await _managed_file_table(self.prisma_client).upsert( where={"unified_file_id": file_id}, data={"create": db_data, "update": update_data}, ) verbose_logger.debug(f"LiteLLM Managed File object with id={file_id} stored in db: {result}") async def store_unified_object_id( self, unified_object_id: str, file_object: Union[LiteLLMBatch, LiteLLMFineTuningJob, "ResponsesAPIResponse"], litellm_parent_otel_span: Optional[Span], model_object_id: str, file_purpose: Literal["batch", "fine-tune", "response"], user_api_key_dict: UserAPIKeyAuth, request_tags: Sequence[str] | None = None, persist_attribution: bool = False, create_if_missing: bool = True, ) -> None: """Persist a managed object row, caching it and upserting it in the DB. persist_attribution is set only by the batch create, which is the one caller that can speak for the creator; it gates the api_key and request_tags columns that CheckBatchCost bills against, so a later poll or retrieve of the same batch cannot record itself as the paying key. Like created_by and team_id, both are written only in the upsert create branch, never on update. create_if_missing is cleared by callers that observe a batch they did not create, such as a poll. They still refresh status and file_object, but a row absent from the table is left absent rather than created with the observer as its creator, because created_by and team_id are written from whoever calls the create branch. """ verbose_logger.info(f"Storing LiteLLM Managed {file_purpose} object with id={unified_object_id} in cache") litellm_managed_object = LiteLLM_ManagedObjectTable( unified_object_id=unified_object_id, model_object_id=model_object_id, file_purpose=file_purpose, file_object=file_object, ) await self.internal_usage_cache.async_set_cache( key=unified_object_id, value=litellm_managed_object.model_dump(), litellm_parent_otel_span=litellm_parent_otel_span, ) from prisma import Json api_key = user_api_key_dict.api_key or None attribution_columns = ( { **({"api_key": api_key} if api_key is not None else {}), **({"request_tags": Json(list(request_tags))} if request_tags else {}), } if persist_attribution else {} ) # FIX: Update status and file_object on every operation to keep state in sync update_columns: Final = { "file_object": file_object.model_dump_json(), "status": file_object.status, "updated_by": user_api_key_dict.user_id, } if not create_if_missing: await _managed_object_table(self.prisma_client).update_many( where={"unified_object_id": unified_object_id}, data=update_columns, ) return await _managed_object_table(self.prisma_client).upsert( where={"unified_object_id": unified_object_id}, data={ "create": { "unified_object_id": unified_object_id, "file_object": file_object.model_dump_json(), "model_object_id": model_object_id, "file_purpose": file_purpose, "created_by": user_api_key_dict.user_id, "team_id": user_api_key_dict.team_id, "updated_by": user_api_key_dict.user_id, "status": file_object.status, **attribution_columns, }, "update": update_columns, }, ) async def get_unified_file_id( self, file_id: str, litellm_parent_otel_span: Optional[Span] = None ) -> Optional[LiteLLM_ManagedFileTable]: ## CHECK CACHE result = cast( Optional[dict], await self.internal_usage_cache.async_get_cache( key=file_id, litellm_parent_otel_span=litellm_parent_otel_span, ), ) if result: return LiteLLM_ManagedFileTable.model_validate(result) ## CHECK DB db_object = await _managed_file_table(self.prisma_client).find_first(where={"unified_file_id": file_id}) if db_object: return LiteLLM_ManagedFileTable.model_validate(db_object.model_dump()) return None async def delete_unified_file_id( self, file_id: str, litellm_parent_otel_span: Optional[Span] = None ) -> OpenAIFileObject: ## get old value initial_value = await _managed_file_table(self.prisma_client).find_first(where={"unified_file_id": file_id}) if initial_value is None: raise Exception(f"LiteLLM Managed File object with id={file_id} not found") ## delete old value await self.internal_usage_cache.async_set_cache( key=file_id, value=None, litellm_parent_otel_span=litellm_parent_otel_span, ) await _managed_file_table(self.prisma_client).delete(where={"unified_file_id": file_id}) return initial_value.file_object async def can_user_call_unified_file_id(self, unified_file_id: str, user_api_key_dict: UserAPIKeyAuth) -> bool: managed_file = await _managed_file_table(self.prisma_client).find_first( where={"unified_file_id": unified_file_id} ) if managed_file: return can_access_resource( user_api_key_dict=user_api_key_dict, created_by=managed_file.created_by, resource_team_id=managed_file.team_id, ) raise HTTPException( status_code=404, detail=f"File not found: {unified_file_id}", ) async def can_user_call_unified_object_id(self, unified_object_id: str, user_api_key_dict: UserAPIKeyAuth) -> bool: managed_object = await _managed_object_table(self.prisma_client).find_first( where={"unified_object_id": unified_object_id} ) if managed_object: return can_access_resource( user_api_key_dict=user_api_key_dict, created_by=managed_object.created_by, resource_team_id=managed_object.team_id, ) raise HTTPException( status_code=404, detail=f"Object not found: {unified_object_id}", ) async def list_user_batches( self, user_api_key_dict: UserAPIKeyAuth, limit: Optional[int] = None, after: Optional[str] = None, provider: Optional[str] = None, target_model_names: Optional[str] = None, llm_router: Optional[Router] = None, ) -> Dict[str, object]: # Provider filtering is not supported for managed batches # This is because the encoded object ids stored in the managed objects table do not contain the provider information # To support provider filtering, we would need to store the provider information in the encoded object ids if provider: raise ProxyException( message="Filtering by 'provider' is not supported when using managed batches.", type="invalid_request_error", param="provider", code=400, ) # Model name filtering is not supported for managed batches # This is because the encoded object ids stored in the managed objects table do not contain the model name # A hash of the model name + litellm_params for the model name is encoded as the model id. This is not sufficient to reliably map the target model names to the model ids. if target_model_names: raise ProxyException( message="Filtering by 'target_model_names' is not supported when using managed batches.", type="invalid_request_error", param="target_model_names", code=400, ) if limit == 0: return build_list_page([]) owner_filter = build_owner_filter(user_api_key_dict) if owner_filter is None: return build_list_page([]) where_clause: Dict[str, object] = {"file_purpose": "batch", **owner_filter} if after: cursor_row = await _managed_object_table(self.prisma_client).find_first( where={**where_clause, "unified_object_id": after} ) if cursor_row is None: raise HTTPException( status_code=400, detail=f"Invalid 'after' cursor: no batch found with id '{after}'.", ) page_size: Final = min(limit or 20, 100) cursor_args: _CursorPageArgs = {"cursor": {"unified_object_id": after}, "skip": 1} if after else {} batches = await _managed_object_table(self.prisma_client).find_many( where=where_clause, take=page_size + 1, order=[{"created_at": "desc"}, {"unified_object_id": "desc"}], **cursor_args, ) has_more = len(batches) > page_size parsed_rows: Final = tuple( (row, batch_obj) for row in batches[:page_size] if (batch_obj := _parse_managed_batch_row(row)) is not None ) unified_id_by_raw_id: Final = await map_raw_file_ids_to_unified( raw_file_ids=frozenset( file_id for _, batch_obj in parsed_rows for file_id in (batch_obj.input_file_id, batch_obj.output_file_id, batch_obj.error_file_id) if file_id and not _is_base64_encoded_unified_file_id(file_id) ), prisma_client=self.prisma_client, ) resolved_batches: Final = [ await self._resolve_listed_batch( row=row, batch_obj=batch_obj, unified_id_by_raw_id=unified_id_by_raw_id, user_api_key_dict=user_api_key_dict, ) for row, batch_obj in parsed_rows ] return build_list_page( [batch_obj for batch_obj in resolved_batches if batch_obj is not None], has_more=has_more, ) async def _resolve_listed_batch( self, row: "PrismaManagedObjectRow", batch_obj: LiteLLMBatch, unified_id_by_raw_id: Mapping[str, str], user_api_key_dict: UserAPIKeyAuth, ) -> Optional[LiteLLMBatch]: apply_unified_file_ids(batch_obj, unified_id_by_raw_id) try: await ensure_batch_response_managed_file_ids( response=batch_obj, managed_files_obj=self, prisma_client=self.prisma_client, verbose_proxy_logger=verbose_logger, user_api_key_dict=user_api_key_dict, db_batch_object=row, unified_batch_id=_is_base64_encoded_unified_file_id(row.unified_object_id), ) except Exception as e: verbose_logger.warning(f"Failed to resolve managed file ids for batch {row.unified_object_id}: {e}") return None return batch_obj async def get_user_created_file_ids( self, user_api_key_dict: UserAPIKeyAuth, model_object_ids: List[str] ) -> List[OpenAIFileObject]: """ Get all file ids the caller is allowed to see for a list of model object ids. Service-account keys (no user_id) are scoped to their team via ``team_id``; admins see all matches. Returns: - List of OpenAIFileObject's """ owner_filter = build_owner_filter(user_api_key_dict) if owner_filter is None: return [] file_ids = await _managed_file_table(self.prisma_client).find_many( where={ **owner_filter, "flat_model_file_ids": {"hasSome": model_object_ids}, } ) return [ parsed_file_object.model_copy(update={"id": row.unified_file_id}) for row in file_ids if (parsed_file_object := _parse_managed_file_object(row.file_object, row.unified_file_id)) is not None ] async def check_managed_file_id_access(self, data: Dict, user_api_key_dict: UserAPIKeyAuth) -> bool: retrieve_file_id = cast(Optional[str], data.get("file_id")) potential_file_id = _is_base64_encoded_unified_file_id(retrieve_file_id) if retrieve_file_id else False if potential_file_id and retrieve_file_id: if await self.can_user_call_unified_file_id(retrieve_file_id, user_api_key_dict): return True else: raise HTTPException( status_code=403, detail=f"User {user_api_key_dict.user_id} does not have access to the file {retrieve_file_id}", ) return False async def check_file_ids_access(self, file_ids: List[str], user_api_key_dict: UserAPIKeyAuth) -> None: """ Check if the user has access to a list of file IDs. Only checks managed (unified) file IDs. Args: file_ids: List of file IDs to check access for user_api_key_dict: User API key authentication details Raises: HTTPException: If user doesn't have access to any of the files """ for file_id in file_ids: is_unified_file_id = _is_base64_encoded_unified_file_id(file_id) if is_unified_file_id: if not await self.can_user_call_unified_file_id(file_id, user_api_key_dict): raise HTTPException( status_code=403, detail=f"User {user_api_key_dict.user_id} does not have access to the file {file_id}", ) async def async_pre_call_hook( self, user_api_key_dict: UserAPIKeyAuth, cache: DualCache, data: Dict, call_type: CallTypesLiteral, ) -> Union[Exception, str, Dict, None]: """ - Detect litellm_proxy/ file_id - add dictionary of mappings of litellm_proxy/ file_id -> provider_file_id => {litellm_proxy/file_id: {"model_id": id, "file_id": provider_file_id}} """ ### HANDLE FILE ACCESS ### - ensure user has access to the file if ( call_type == CallTypes.afile_content.value or call_type == CallTypes.afile_delete.value or call_type == CallTypes.afile_retrieve.value or call_type == CallTypes.afile_content.value ): await self.check_managed_file_id_access(data, user_api_key_dict) ### HANDLE TRANSFORMATIONS ### # Check both completion and acompletion call types is_completion_call = call_type == CallTypes.completion.value or call_type == CallTypes.acompletion.value if is_completion_call: messages = data.get("messages") model = data.get("model", "") if messages: file_ids = self.get_file_ids_from_messages(messages) if file_ids: # Check user has access to all managed files await self.check_file_ids_access(file_ids, user_api_key_dict) # Check if any files are stored in storage backends and need base64 conversion # This is needed for Vertex AI/Gemini which requires base64 content is_vertex_ai = model and ("vertex_ai" in model or "gemini" in model.lower()) if is_vertex_ai: await self._convert_storage_files_to_base64( messages=messages, file_ids=file_ids, litellm_parent_otel_span=user_api_key_dict.parent_otel_span, ) model_file_id_mapping = await self.get_model_file_id_mapping( file_ids, user_api_key_dict.parent_otel_span ) data["model_file_id_mapping"] = model_file_id_mapping elif call_type == CallTypes.aresponses.value or call_type == CallTypes.responses.value: # Handle managed files in responses API input and tools file_ids = [] # Extract file IDs from input parameter input_data = data.get("input") if input_data: file_ids.extend(self.get_file_ids_from_responses_input(input_data)) # Extract file IDs from tools parameter (e.g., code_interpreter container) tools = data.get("tools") if tools: file_ids.extend(self.get_file_ids_from_responses_tools(tools)) if file_ids: # Check user has access to all managed files await self.check_file_ids_access(file_ids, user_api_key_dict) model_file_id_mapping = await self.get_model_file_id_mapping( file_ids, user_api_key_dict.parent_otel_span ) data["model_file_id_mapping"] = model_file_id_mapping # Check access for file_search vector_store_ids if tools: unified_vs_ids = self.get_vector_store_ids_from_file_search_tools(tools) if unified_vs_ids: await self.check_vector_store_ids_access(unified_vs_ids, user_api_key_dict) elif call_type == CallTypes.afile_content.value: retrieve_file_id = cast(Optional[str], data.get("file_id")) potential_file_id = _is_base64_encoded_unified_file_id(retrieve_file_id) if retrieve_file_id else False if potential_file_id and "llm_output_file_id," in potential_file_id: model_id = self.get_model_id_from_unified_file_id(potential_file_id) if model_id: data["model"] = model_id data["file_id"] = self.get_output_file_id_from_unified_file_id(potential_file_id) elif call_type == CallTypes.acreate_batch.value: input_file_id = cast(Optional[str], data.get("input_file_id")) if input_file_id: model_file_id_mapping = await self.get_model_file_id_mapping( [input_file_id], user_api_key_dict.parent_otel_span ) data["model_file_id_mapping"] = model_file_id_mapping elif ( call_type == CallTypes.aretrieve_batch.value or call_type == CallTypes.acancel_batch.value or call_type == CallTypes.acancel_fine_tuning_job.value or call_type == CallTypes.aretrieve_fine_tuning_job.value ): accessor_key: Optional[str] = None retrieve_object_id: Optional[str] = None if call_type == CallTypes.aretrieve_batch.value or call_type == CallTypes.acancel_batch.value: accessor_key = "batch_id" elif ( call_type == CallTypes.acancel_fine_tuning_job.value or call_type == CallTypes.aretrieve_fine_tuning_job.value ): accessor_key = "fine_tuning_job_id" if accessor_key: retrieve_object_id = cast(Optional[str], data.get(accessor_key)) potential_llm_object_id = ( _is_base64_encoded_unified_file_id(retrieve_object_id) if retrieve_object_id else False ) if potential_llm_object_id and retrieve_object_id: ## VALIDATE USER HAS ACCESS TO THE OBJECT ## if not await self.can_user_call_unified_object_id(retrieve_object_id, user_api_key_dict): raise HTTPException( status_code=403, detail=f"User {user_api_key_dict.user_id} does not have access to the object {retrieve_object_id}", ) ## for managed batch id - get the model id potential_model_id = get_model_id_from_unified_batch_id(potential_llm_object_id) if potential_model_id is None: raise Exception( f"LiteLLM Managed {accessor_key} with id={retrieve_object_id} is invalid - does not contain encoded model_id." ) data["model"] = potential_model_id data[accessor_key] = get_batch_id_from_unified_batch_id(potential_llm_object_id) elif call_type == CallTypes.acreate_fine_tuning_job.value: input_file_id = cast(Optional[str], data.get("training_file")) if input_file_id: model_file_id_mapping = await self.get_model_file_id_mapping( [input_file_id], user_api_key_dict.parent_otel_span ) return data async def async_filter_deployments( self, model: str, healthy_deployments: List, messages: Optional[List[AllMessageValues]], request_kwargs: Optional[Dict] = None, parent_otel_span: Optional[Span] = None, ) -> List[Dict]: if request_kwargs is None: return healthy_deployments input_file_id = cast(Optional[str], request_kwargs.get("input_file_id")) model_file_id_mapping = cast( Optional[Dict[str, Dict[str, str]]], request_kwargs.get("model_file_id_mapping"), ) allowed_model_ids = [] if input_file_id and model_file_id_mapping: model_id_dict = model_file_id_mapping.get(input_file_id, {}) allowed_model_ids = list(model_id_dict.keys()) if len(allowed_model_ids) == 0: return healthy_deployments return [ deployment for deployment in healthy_deployments if deployment.get("model_info", {}).get("id") in allowed_model_ids ] async def async_pre_call_deployment_hook( self, kwargs: Dict[str, Any], call_type: Optional[CallTypes] ) -> Optional[dict]: """ Allow modifying the request just before it's sent to the deployment. """ accessor_key: Optional[str] = None if call_type and call_type == CallTypes.acreate_batch: accessor_key = "input_file_id" elif call_type and call_type == CallTypes.acreate_fine_tuning_job: accessor_key = "training_file" else: return kwargs if accessor_key: input_file_id = cast(Optional[str], kwargs.get(accessor_key)) model_file_id_mapping = cast(Optional[Dict[str, Dict[str, str]]], kwargs.get("model_file_id_mapping")) # model_info may be at top-level or nested under litellm_metadata # (batch/file operations use litellm_metadata) model_id = cast(Optional[str], kwargs.get("model_info", {}).get("id", None)) if model_id is None: model_id = cast( Optional[str], kwargs.get("litellm_metadata", {}).get("model_info", {}).get("id", None), ) mapped_file_id: Optional[str] = None if input_file_id and model_file_id_mapping and model_id: mapped_file_id = model_file_id_mapping.get(input_file_id, {}).get(model_id, None) if mapped_file_id: kwargs[accessor_key] = mapped_file_id return kwargs def get_file_ids_from_messages(self, messages: List[AllMessageValues]) -> List[str]: """ Gets file ids from messages """ file_ids = [] for message in messages: if message.get("role") == "user": content = message.get("content") if content: if isinstance(content, str): continue for c in content: if c.get("type") == "file": file_object = cast(ChatCompletionFileObject, c) file_object_file_field = file_object["file"] file_id = file_object_file_field.get("file_id") if file_id: file_ids.append(file_id) return file_ids def get_file_ids_from_responses_input(self, input: Union[str, List[Dict[str, Any]]]) -> List[str]: """ Gets file ids from responses API input. The input can be: - A string (no files) - A list of input items, where each item can have: - type: "input_file" with file_id - content: a list that can contain items with type: "input_file" and file_id """ file_ids: List[str] = [] if isinstance(input, str): return file_ids if not isinstance(input, list): return file_ids for item in input: if not isinstance(item, dict): continue # Check for direct input_file type if item.get("type") == "input_file": file_id = item.get("file_id") if file_id: file_ids.append(file_id) # Check for input_file in content array content = item.get("content") if isinstance(content, list): for content_item in content: if isinstance(content_item, dict) and content_item.get("type") == "input_file": file_id = content_item.get("file_id") if file_id: file_ids.append(file_id) return file_ids def get_file_ids_from_responses_tools(self, tools: List[Dict[str, object]]) -> List[str]: """ Gets file ids from responses API tools parameter. The tools can contain code_interpreter with container.file_ids: [ { "type": "code_interpreter", "container": {"type": "auto", "file_ids": ["file-123", "file-456"]} } ] """ file_ids: List[str] = [] if not isinstance(tools, list): return file_ids for tool in tools: if not isinstance(tool, dict): continue # Check for code_interpreter with container file_ids if tool.get("type") == "code_interpreter": container = tool.get("container") if isinstance(container, dict): container_file_ids = container.get("file_ids") if isinstance(container_file_ids, list): for file_id in container_file_ids: if isinstance(file_id, str): file_ids.append(file_id) return file_ids def get_vector_store_ids_from_file_search_tools(self, tools: List[Dict[str, object]]) -> List[str]: """ Extract unified vector_store_ids from file_search tools. Only returns IDs that are LiteLLM-managed (base64 unified IDs). Native provider IDs are skipped — they have no LiteLLM access record. """ from litellm.llms.base_llm.managed_resources.utils import ( is_base64_encoded_unified_id, ) vs_ids: List[str] = [] if not isinstance(tools, list): return vs_ids for tool in tools: if not isinstance(tool, dict) or tool.get("type") != "file_search": continue vector_store_ids = tool.get("vector_store_ids") if not isinstance(vector_store_ids, list): continue for vs_id in vector_store_ids: if isinstance(vs_id, str) and is_base64_encoded_unified_id(vs_id): vs_ids.append(vs_id) return vs_ids async def check_vector_store_ids_access( self, vector_store_ids: List[str], user_api_key_dict: UserAPIKeyAuth, ) -> None: """ Verify the caller's team can access each LiteLLM-managed vector store. Batch-fetches vector stores from DB and checks team_id. Raises HTTPException(403) on the first access violation. Non-managed (native) IDs should already be filtered out before calling this. """ from litellm.llms.base_llm.managed_resources.utils import ( extract_unified_uuid_from_unified_id, ) from litellm.proxy.auth.auth_checks import ( get_managed_vector_store_rows_by_uuids, ) from litellm.proxy.proxy_server import ( prisma_client, proxy_logging_obj, user_api_key_cache, ) if not vector_store_ids or prisma_client is None: return # Map each unified ID to its internal UUID for a single batch DB fetch uuid_to_unified: Dict[str, str] = {} for vs_id in vector_store_ids: uuid = extract_unified_uuid_from_unified_id(vs_id) if uuid: uuid_to_unified[uuid] = vs_id if not uuid_to_unified: return rows = await get_managed_vector_store_rows_by_uuids( uuids=list(uuid_to_unified.keys()), prisma_client=prisma_client, user_api_key_cache=user_api_key_cache, proxy_logging_obj=proxy_logging_obj, ) found_uuids = {row.vector_store_id for row in rows} for uuid, original_id in uuid_to_unified.items(): if uuid not in found_uuids: raise HTTPException( status_code=403, detail=f"Vector store '{original_id}' not found or access denied.", ) caller_team_id = user_api_key_dict.team_id for row in rows: vs_team_id = getattr(row, "team_id", None) if vs_team_id is not None and vs_team_id != caller_team_id: raise HTTPException( status_code=403, detail=( f"Team '{caller_team_id}' does not have access to vector " f"store '{row.vector_store_id}'. The store belongs to team " f"'{vs_team_id}'." ), ) async def get_model_file_id_mapping(self, file_ids: List[str], litellm_parent_otel_span: Span) -> dict: """ Get model-specific file IDs for a list of proxy file IDs. Returns a dictionary mapping litellm_proxy/ file_id -> model_id -> model_file_id 1. Get all the litellm_proxy/ file_ids from the messages 2. For each file_id, search for cache keys matching the pattern file_id:* 3. Return a dictionary of mappings of litellm_proxy/ file_id -> model_id -> model_file_id Example: { "litellm_proxy/file_id": { "model_id": "model_file_id" } } """ file_id_mapping: Dict[str, Dict[str, str]] = {} litellm_managed_file_ids = [] for file_id in file_ids: ## CHECK IF FILE ID IS MANAGED BY LITELM is_base64_unified_file_id = _is_base64_encoded_unified_file_id(file_id) if is_base64_unified_file_id: litellm_managed_file_ids.append(file_id) if litellm_managed_file_ids: # Get all cache keys matching the pattern file_id:* for file_id in litellm_managed_file_ids: # Search for any cache key starting with this file_id unified_file_object = await self.get_unified_file_id(file_id, litellm_parent_otel_span) if unified_file_object: file_id_mapping[file_id] = unified_file_object.model_mappings return file_id_mapping async def create_file_for_each_model( self, llm_router: Optional[Router], _create_file_request: CreateFileRequest, target_model_names_list: List[str], litellm_parent_otel_span: Span, ) -> List[OpenAIFileObject]: if llm_router is None: raise Exception("LLM Router not initialized. Ensure models added to proxy.") responses = [] for model in target_model_names_list: individual_response = await llm_router.acreate_file(model=model, **_create_file_request) responses.append(individual_response) return responses async def acreate_file( self, create_file_request: CreateFileRequest, llm_router: Router, target_model_names_list: List[str], litellm_parent_otel_span: Span, user_api_key_dict: UserAPIKeyAuth, ) -> OpenAIFileObject: responses = await self.create_file_for_each_model( llm_router=llm_router, _create_file_request=create_file_request, target_model_names_list=target_model_names_list, litellm_parent_otel_span=litellm_parent_otel_span, ) response = await _PROXY_LiteLLMManagedFiles.return_unified_file_id( file_objects=responses, create_file_request=create_file_request, internal_usage_cache=self.internal_usage_cache, litellm_parent_otel_span=litellm_parent_otel_span, target_model_names_list=target_model_names_list, ) ## STORE MODEL MAPPINGS IN DB model_mappings: Dict[str, str] = {} for file_object in responses: model_file_id_mapping = file_object._hidden_params.get("model_file_id_mapping") if model_file_id_mapping and isinstance(model_file_id_mapping, dict): model_mappings.update(model_file_id_mapping) await self.store_unified_file_id( file_id=response.id, file_object=response, litellm_parent_otel_span=litellm_parent_otel_span, model_mappings=model_mappings, user_api_key_dict=user_api_key_dict, ) # Emit Prometheus metrics for managed file creation prom_logger = self._get_prometheus_logger() if prom_logger: first_model = target_model_names_list[0] if target_model_names_list else None first_provider = "" if responses: first_provider = getattr(responses[0], "_hidden_params", {}).get("custom_llm_provider") or "" prom_logger.record_managed_file_created( model=first_model or "", api_provider=first_provider, user=user_api_key_dict.user_id or "", user_email=getattr(user_api_key_dict, "user_email", None) or "", api_key_alias=user_api_key_dict.key_alias or "", ) if response.bytes and response.bytes > 0: prom_logger.record_managed_file_size( size_bytes=response.bytes, purpose=response.purpose or "batch", file_type="input", model=first_model, api_provider=first_provider, user=user_api_key_dict.user_id, ) return response @staticmethod async def return_unified_file_id( file_objects: List[OpenAIFileObject], create_file_request: CreateFileRequest, internal_usage_cache: InternalUsageCache, litellm_parent_otel_span: Span, target_model_names_list: List[str], ) -> OpenAIFileObject: ## GET THE FILE TYPE FROM THE CREATE FILE REQUEST _, file_type = extract_file_metadata(create_file_request["file"]) output_file_id = file_objects[0].id model_id = file_objects[0]._hidden_params.get("model_id") unified_file_id = SpecialEnums.LITELLM_MANAGED_FILE_COMPLETE_STR.value.format( file_type, str(uuid.uuid4()), ",".join(target_model_names_list), output_file_id, model_id, ) # Convert to URL-safe base64 and strip padding base64_unified_file_id = base64.urlsafe_b64encode(unified_file_id.encode()).decode().rstrip("=") ## CREATE RESPONSE OBJECT response = OpenAIFileObject( id=base64_unified_file_id, object="file", purpose=create_file_request["purpose"], created_at=file_objects[0].created_at, bytes=file_objects[0].bytes, filename=file_objects[0].filename, status="uploaded", expires_at=file_objects[0].expires_at, ) return response def get_unified_generic_response_id(self, model_id: str, generic_response_id: str) -> str: unified_generic_response_id = SpecialEnums.LITELLM_MANAGED_GENERIC_RESPONSE_COMPLETE_STR.value.format( model_id, generic_response_id ) return base64.urlsafe_b64encode(unified_generic_response_id.encode()).decode().rstrip("=") def get_unified_batch_id(self, batch_id: str, model_id: str) -> str: unified_batch_id = SpecialEnums.LITELLM_MANAGED_BATCH_COMPLETE_STR.value.format(model_id, batch_id) return base64.urlsafe_b64encode(unified_batch_id.encode()).decode().rstrip("=") def get_unified_output_file_id(self, output_file_id: str, model_id: str, model_name: Optional[str]) -> str: deterministic_uuid: Final = uuid5(uuid5(NAMESPACE_URL, model_id), output_file_id) unified_output_file_id = SpecialEnums.LITELLM_MANAGED_FILE_COMPLETE_STR.value.format( "application/json", str(deterministic_uuid), model_name or "", output_file_id, model_id, ) return base64.urlsafe_b64encode(unified_output_file_id.encode()).decode().rstrip("=") def get_model_id_from_unified_file_id(self, file_id: str) -> str: return file_id.split("llm_output_file_model_id,")[1].split(";")[0] def get_output_file_id_from_unified_file_id(self, file_id: str) -> str: marker = "llm_output_file_id," if marker not in file_id: raise ValueError(f"Unified id does not contain {marker!r}: {file_id[:80]!r}") return file_id.split(marker, 1)[1].split(";")[0] async def async_post_call_success_hook( self, data: Dict, user_api_key_dict: UserAPIKeyAuth, response: LLMResponseTypes ) -> LLMResponseTypes: if isinstance(response, LiteLLMBatch): ## Check if unified_file_id is in the response unified_file_id = response._hidden_params.get("unified_file_id") # managed file id unified_batch_id = response._hidden_params.get("unified_batch_id") # managed batch id is_batch_create: Final = unified_file_id is not None model_id = cast(Optional[str], response._hidden_params.get("model_id")) model_name = cast(Optional[str], response._hidden_params.get("model_name")) resolved_model_name = resolve_managed_output_file_model_name( unified_input_file_id=unified_file_id if isinstance(unified_file_id, str) else response.input_file_id, fallback_model_name=model_name, ) original_response_id = response.id if (unified_batch_id or unified_file_id) and model_id: response.id = self.get_unified_batch_id(batch_id=response.id, model_id=model_id) # Handle both output_file_id and error_file_id for file_attr in ["output_file_id", "error_file_id"]: file_id_value = getattr(response, file_attr, None) if file_id_value and model_id: decoded_output_file_id = _is_base64_encoded_unified_file_id(file_id_value) if decoded_output_file_id and "llm_output_file_id," in decoded_output_file_id: provider_file_id = self.get_output_file_id_from_unified_file_id(decoded_output_file_id) unified_file_id = file_id_value elif decoded_output_file_id: verbose_logger.warning( f"Skipping {file_attr}={file_id_value!r}: unified id is not a managed file output id" ) continue else: provider_file_id = file_id_value unified_file_id = self.get_unified_output_file_id( output_file_id=provider_file_id, model_id=model_id, model_name=resolved_model_name, ) setattr(response, file_attr, unified_file_id) # Use llm_router credentials when available. Without credentials, # Azure and other auth-required providers return 500/401. file_object = None try: # Import module and use getattr for better testability with mocks import litellm.proxy.proxy_server as proxy_server_module _llm_router = getattr(proxy_server_module, "llm_router", None) if _llm_router is not None and model_id: _creds = _llm_router.get_deployment_credentials_with_provider(model_id) or {} file_object = await litellm.afile_retrieve( file_id=provider_file_id, **_creds, ) else: file_object = await litellm.afile_retrieve( custom_llm_provider=model_name.split("/")[0] if model_name and "/" in model_name else "openai", # type: ignore[arg-type] file_id=provider_file_id, ) verbose_logger.debug( f"Successfully retrieved file object for {file_attr}={provider_file_id}" ) except Exception as e: verbose_logger.warning( f"Failed to retrieve file object for {file_attr}={provider_file_id}: {str(e)}. Storing with None and will fetch on-demand." ) await self.store_unified_file_id( file_id=unified_file_id, file_object=file_object, litellm_parent_otel_span=user_api_key_dict.parent_otel_span, model_mappings={model_id: provider_file_id}, user_api_key_dict=user_api_key_dict, ) request_metadata: Final = data.get("litellm_metadata") await self.store_unified_object_id( unified_object_id=response.id, file_object=response, litellm_parent_otel_span=user_api_key_dict.parent_otel_span, model_object_id=original_response_id, file_purpose="batch", user_api_key_dict=user_api_key_dict, request_tags=request_tags_from_metadata(request_metadata if isinstance(request_metadata, dict) else {}), persist_attribution=is_batch_create, ) # Only record batch creation metric on actual create (not retrieve/cancel). # unified_file_id in _hidden_params is only set by the create_batch endpoint. original_unified_file_id = response._hidden_params.get("unified_file_id") if original_unified_file_id: prom_logger = self._get_prometheus_logger() if prom_logger: batch_provider = "" if model_name: try: from litellm.litellm_core_utils.get_llm_provider_logic import ( get_llm_provider, ) _, batch_provider, _, _ = get_llm_provider(model=model_name) except Exception: if "/" in model_name: batch_provider = model_name.split("/")[0] prom_logger.record_managed_batch_created( model=model_name or "", api_provider=batch_provider, user=user_api_key_dict.user_id or "", user_email=getattr(user_api_key_dict, "user_email", None) or "", api_key_alias=user_api_key_dict.key_alias or "", ) elif isinstance(response, LiteLLMFineTuningJob): ## Check if unified_file_id is in the response unified_file_id = response._hidden_params.get("unified_file_id") # managed file id unified_finetuning_job_id = response._hidden_params.get( "unified_finetuning_job_id" ) # managed finetuning job id model_id = cast(Optional[str], response._hidden_params.get("model_id")) model_name = cast(Optional[str], response._hidden_params.get("model_name")) original_response_id = response.id if (unified_file_id or unified_finetuning_job_id) and model_id: response.id = self.get_unified_generic_response_id(model_id=model_id, generic_response_id=response.id) await self.store_unified_object_id( unified_object_id=response.id, file_object=response, litellm_parent_otel_span=user_api_key_dict.parent_otel_span, model_object_id=original_response_id, file_purpose="fine-tune", user_api_key_dict=user_api_key_dict, ) elif isinstance(response, AsyncCursorPage): """ For listing files, filter for the ones created by the user """ ## check if file object if hasattr(response, "data") and isinstance(response.data, list): if all(isinstance(file_object, FileObject) for file_object in response.data): ## Get all file id's ## Check which file id's were created by the user ## Filter the response to only include the files created by the user ## Return the filtered response file_ids = [ file_object.id for file_object in cast(List[FileObject], response.data) # type: ignore ] user_created_file_ids = await self.get_user_created_file_ids(user_api_key_dict, file_ids) ## Filter the response to only include the files created by the user response.data = user_created_file_ids # type: ignore self._scope_list_page_cursors(response, user_created_file_ids) return response return response return response @staticmethod def _scope_list_page_cursors(response: AsyncCursorPage, data: List[OpenAIFileObject]) -> None: """Rebuild ``first_id`` / ``last_id`` from the caller-scoped page. The upstream cursors point at rows that were just filtered out, so leaving them in place discloses other callers' file ids. ``has_more`` is always cleared because ``after`` is never forwarded upstream, so no further page is reachable through the proxy. """ if hasattr(response, "first_id"): response.first_id = data[0].id if data else None if hasattr(response, "last_id"): response.last_id = data[-1].id if data else None if hasattr(response, "has_more"): response.has_more = False async def afile_retrieve( self, file_id: str, litellm_parent_otel_span: Optional[Span], llm_router: Optional[Router] = None ) -> OpenAIFileObject: stored_file_object = await self.get_unified_file_id(file_id, litellm_parent_otel_span) # Case 1 : This is not a managed file if not stored_file_object: raise Exception(f"LiteLLM Managed File object with id={file_id} not found") # Case 2: Managed file and the file object exists in the database # The stored file_object has the raw provider ID. Replace with the unified ID # so callers see a consistent ID (matching Case 3 which does response.id = file_id). if stored_file_object and stored_file_object.file_object: # Use model_copy to ensure the ID update persists (Pydantic v2 compatibility) response = stored_file_object.file_object.model_copy(update={"id": file_id}) return response # Case 3: Managed file exists in the database but not the file object (for. e.g the batch task might not have run) # So we fetch the file object from the provider. We deliberately do not store the result to avoid interfering with batch cost tracking code. if not llm_router: raise Exception( f"LiteLLM Managed File object with id={file_id} has no file_object " f"and llm_router is required to fetch from provider" ) try: model_id, model_file_id = next(iter(stored_file_object.model_mappings.items())) credentials = llm_router.get_deployment_credentials_with_provider(model_id) or {} response = await litellm.afile_retrieve(file_id=model_file_id, **credentials) response.id = file_id # Replace with unified ID return response except Exception as e: raise Exception(f"Failed to retrieve file {file_id} from provider: {str(e)}") from e async def afile_list( self, purpose: Optional[str], litellm_parent_otel_span: Optional[Span], user_api_key_dict: UserAPIKeyAuth, limit: Optional[int] = None, after: Optional[str] = None, **data: Dict, ) -> FileListPage: """List the managed files the caller owns, newest first. Pagination is keyset based on ``unified_file_id`` so a key that owns every file on the proxy still reads one bounded page at a time. ``purpose`` is applied after parsing, because the managed file table keeps it inside the ``file_object`` blob instead of a column, and rows whose blob will not parse drop out there too, so a chunk of rows can yield fewer matches than the page holds. Successive chunks are read until the page is full or the caller's rows run out, which keeps ``data`` non-empty while matches remain and its last id usable as the next cursor. A first chunk that fills the page costs one query; once a scan has to continue past it, the chunk widens to ``FILE_LIST_CONTINUATION_CHUNK_SIZE``, so the walk costs one query per that many rows instead of one per page. That bound is per query, not per request: the work is still linear in the rows the caller owns, and a filter matching nothing reads every one of them, with no index covering either the owner filter or the sort. """ validate_file_list_limit(limit) validate_file_list_purpose(purpose) owner_filter: Final = build_owner_filter(user_api_key_dict) if owner_filter is None: return FileListPage(**build_list_page([])) if after: cursor_row = await _managed_file_table(self.prisma_client).find_first( where={**owner_filter, "unified_file_id": after} ) if cursor_row is None: raise ProxyException( message=f"Invalid 'after' cursor: no file found with id '{after}'.", type="invalid_request_error", param="after", code=400, openai_code="invalid_value", ) page_size: Final = min(limit or MAX_FILE_LIST_LIMIT, MAX_FILE_LIST_LIMIT) matches: Final[List[OpenAIFileObject]] = [] cursor_id = after chunk_size = page_size + 1 while len(matches) <= page_size: cursor_args: _CursorPageArgs = {"cursor": {"unified_file_id": cursor_id}, "skip": 1} if cursor_id else {} chunk = await _managed_file_table(self.prisma_client).find_many( where=owner_filter, take=chunk_size, order=[{"created_at": "desc"}, {"unified_file_id": "desc"}], **cursor_args, ) matches.extend( parsed_file_object.model_copy(update={"id": row.unified_file_id}) for row in chunk if (parsed_file_object := _parse_managed_file_object(row.file_object, row.unified_file_id)) is not None and (purpose is None or parsed_file_object.purpose == purpose) ) if len(chunk) < chunk_size: break cursor_id = chunk[-1].unified_file_id chunk_size = max(chunk_size, FILE_LIST_CONTINUATION_CHUNK_SIZE) return FileListPage(**build_list_page(matches[:page_size], has_more=len(matches) > page_size)) def _is_batch_polling_enabled(self) -> bool: """ Check if batch cost tracking is actually enabled and running. Returns: bool: True if batch cost tracking is active, False otherwise """ try: # Import here to avoid circular dependencies import litellm.proxy.proxy_server as proxy_server_module # Check if the scheduler has the batch cost checking job registered scheduler = getattr(proxy_server_module, "scheduler", None) if scheduler is None: return False # Check if the check_batch_cost_job exists in the scheduler try: job = scheduler.get_job("check_batch_cost_job") if job is not None: return True except Exception: # Job not found or scheduler doesn't support get_job pass return False except Exception as e: verbose_logger.warning(f"Error checking batch polling configuration: {e}. Assuming disabled.") return False async def _get_batches_referencing_file(self, file_id: str) -> List[Dict[str, object]]: """ Find batches that reference this file and still need cost tracking. Find batches that are in non-terminal state and have not yet been processed by CheckBatchCost. Args: file_id: The unified file ID to check Returns: List of batch objects referencing this file in non-terminal state (max 10 for error message display) """ # Prepare list of file IDs to check (both unified and provider IDs) file_ids_to_check = [file_id] # Get model-specific file IDs for this unified file ID if it's a managed file try: model_file_id_mapping = await self.get_model_file_id_mapping([file_id], litellm_parent_otel_span=None) if model_file_id_mapping and file_id in model_file_id_mapping: # Add all provider file IDs for this unified file provider_file_ids = list(model_file_id_mapping[file_id].values()) file_ids_to_check.extend(provider_file_ids) except Exception as e: verbose_logger.debug( f"Could not get model file ID mapping for {file_id}: {e}. Will only check unified file ID." ) MAX_MATCHES_TO_RETURN = 10 batches = await self.prisma_client.db.litellm_managedobjecttable.find_many( where={ "file_purpose": "batch", "batch_processed": False, "status": {"not_in": ["failed", "expired", "cancelled"]}, }, take=MAX_MATCHES_TO_RETURN, order={"created_at": "desc"}, ) referencing_batches = [] for batch in batches: try: # Parse the batch file_object to check for file references batch_data = json.loads(batch.file_object) if isinstance(batch.file_object, str) else batch.file_object # Extract file IDs from batch # Batches typically reference the unified file ID in input_file_id # Output and error files are generated by the provider input_file_id = batch_data.get("input_file_id") output_file_id = batch_data.get("output_file_id") error_file_id = batch_data.get("error_file_id") referenced_file_ids = [fid for fid in [input_file_id, output_file_id, error_file_id] if fid] # Check if any referenced file ID matches the file we're trying to delete if any(ref_id in file_ids_to_check for ref_id in referenced_file_ids): referencing_batches.append( { "batch_id": batch.unified_object_id, "status": batch.status, "created_at": batch.created_at, } ) except Exception as e: verbose_logger.warning(f"Error parsing batch object {batch.unified_object_id}: {e}") continue return referencing_batches async def _check_file_deletion_allowed(self, file_id: str) -> None: """ Check if file deletion should be blocked due to batch references. Blocks deletion if: 1. File is referenced by any batch in non-terminal state, AND 2. Batch polling is configured (user wants cost tracking) Args: file_id: The unified file ID to check Raises: HTTPException: If file deletion should be blocked """ # Check if batch polling is enabled if not self._is_batch_polling_enabled(): # Batch polling not configured, allow deletion return # Check if file is referenced by any non-terminal batches referencing_batches = await self._get_batches_referencing_file(file_id) if referencing_batches: # File is referenced by non-terminal batches and polling is enabled MAX_BATCHES_IN_ERROR = 5 # Limit batches shown in error message for readability # Show up to MAX_BATCHES_IN_ERROR in the error message batches_to_show = referencing_batches[:MAX_BATCHES_IN_ERROR] batch_statuses = [f"{b['batch_id']}: {b['status']}" for b in batches_to_show] # Determine the count message count_message = f"{len(referencing_batches)}" if len(referencing_batches) >= 10: # MAX_MATCHES_TO_RETURN from _get_batches_referencing_file count_message = "10+" error_message = ( f"Cannot delete file {file_id}. " f"The file is referenced by {count_message} batch(es) in non-terminal state" ) # Add specific batch details if not too many if len(referencing_batches) <= MAX_BATCHES_IN_ERROR: error_message += f": {', '.join(batch_statuses)}. " else: error_message += f" (showing {MAX_BATCHES_IN_ERROR} most recent): {', '.join(batch_statuses)}. " error_message += ( "To delete this file before complete cost tracking, please delete or cancel the referencing batch(es) first. " "Alternatively, wait for all batches to complete and for cost to be computed (batch_processed=true)." ) # Record blocked deletion metric prom_logger = self._get_prometheus_logger() if prom_logger: prom_logger.record_managed_file_deleted(result="blocked") raise HTTPException( status_code=400, detail=error_message, ) async def afile_delete( self, file_id: str, litellm_parent_otel_span: Optional[Span], llm_router: Router, **data: Dict, ) -> OpenAIFileObject: # Check if file deletion should be blocked due to batch references await self._check_file_deletion_allowed(file_id) # 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 filtered_data = {k: v for k, v in data.items() if k not in ("model", "file_id")} for model_id, model_file_id in specific_model_file_id_mapping.items(): delete_response = await llm_router.afile_delete(model=model_id, file_id=model_file_id, **filtered_data) # type: ignore stored_file_object = 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 stored_file_object 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") async def afile_content( self, file_id: str, litellm_parent_otel_span: Optional[Span], llm_router: Router, **data: Dict, ) -> "HttpxBinaryResponseContent": """ Get the content of a file from first model that has it """ model_file_id_mapping = data.pop("model_file_id_mapping", None) model_file_id_mapping = model_file_id_mapping or await self.get_model_file_id_mapping( [file_id], litellm_parent_otel_span ) specific_model_file_id_mapping = model_file_id_mapping.get(file_id) if specific_model_file_id_mapping: exception_dict = {} for model_id, provider_file_id in specific_model_file_id_mapping.items(): try: # Cloud-storage providers (e.g. Bedrock S3) validate file ids # against the deployment's configured bucket, which they only # trust from this immutable server-side snapshot, never from # request params. credentials = llm_router.get_deployment_credentials_with_provider(model_id=model_id) if credentials is not None: data["_litellm_internal_model_credentials"] = cast(Dict, MappingProxyType(dict(credentials))) else: data.pop("_litellm_internal_model_credentials", None) return await llm_router.afile_content(model=model_id, file_id=provider_file_id, **data) # type: ignore except Exception as e: exception_dict[model_id] = str(e) raise Exception( f"LiteLLM Managed File object with id={file_id} not found. Checked model id's: {specific_model_file_id_mapping.keys()}. Errors: {exception_dict}" ) else: raise Exception(f"LiteLLM Managed File object with id={file_id} not found") async def _convert_storage_files_to_base64( self, messages: List[AllMessageValues], file_ids: List[str], litellm_parent_otel_span: Optional[Span], ) -> None: """ Convert files stored in storage backends to base64 format for Vertex AI/Gemini. This method checks if any managed files are stored in storage backends, downloads them, and converts them to base64 format in the messages. """ # Check each file_id to see if it's stored in a storage backend for file_id in file_ids: # Check if this is a base64 encoded unified file ID decoded_unified_file_id = _is_base64_encoded_unified_file_id(file_id) if not decoded_unified_file_id: continue # Check database for storage backend info # IMPORTANT: The database stores the base64 encoded unified_file_id (not the decoded version) # So we query with the original file_id (which is base64 encoded) db_file = await _managed_file_table(self.prisma_client).find_first(where={"unified_file_id": file_id}) if not db_file or not db_file.storage_backend or not db_file.storage_url: continue # File is stored in a storage backend, download and convert to base64 try: from litellm.llms.base_llm.files.storage_backend_factory import ( get_storage_backend, ) storage_backend_name = db_file.storage_backend storage_url = db_file.storage_url # Get storage backend (uses same env vars as callback) try: storage_backend = get_storage_backend(storage_backend_name) except ValueError as e: verbose_logger.warning( f"Storage backend '{storage_backend_name}' error for file {file_id}: {str(e)}" ) continue file_content = await storage_backend.download_file(storage_url) # Determine content type from file object content_type = self._get_content_type_from_file_object(db_file.file_object) # Convert to base64 base64_data = base64.b64encode(file_content).decode("utf-8") base64_data_uri = f"data:{content_type};base64,{base64_data}" # Update messages to use base64 instead of file_id self._update_messages_with_base64_data(messages, file_id, base64_data_uri, content_type) except Exception as e: verbose_logger.exception(f"Error converting file {file_id} from storage backend to base64: {str(e)}") # Continue with other files even if one fails continue def _get_content_type_from_file_object(self, file_object: Optional[Any]) -> str: """ Determine content type from file object. Uses the MIME type utility for consistent detection and normalization. Args: file_object: The file object from the database (can be dict, JSON string, or None) Returns: str: MIME type (defaults to "application/octet-stream" if cannot be determined) """ # Use utility function for detection content_type = get_content_type_from_file_object(file_object) # Normalize for Gemini/Vertex AI (requires image/jpeg, not image/jpg) content_type = normalize_mime_type_for_provider(content_type, provider="gemini") return content_type def _update_messages_with_base64_data( self, messages: List[AllMessageValues], file_id: str, base64_data_uri: str, content_type: str, ) -> None: """ Update messages to replace file_id with base64 data URI. Args: messages: List of messages to update file_id: The file ID to replace base64_data_uri: The base64 data URI to use as replacement content_type: The MIME type of the file (e.g., "image/jpeg", "application/pdf") """ for message in messages: if message.get("role") == "user": content = message.get("content") if content and isinstance(content, list): for element in content: if element.get("type") == "file": file_element = cast(ChatCompletionFileObject, element) file_element_file = file_element.get("file", {}) if file_element_file.get("file_id") == file_id: # Replace file_id with base64 data file_element_file["file_data"] = base64_data_uri # Set format to help Gemini determine mime type file_element_file["format"] = content_type # Remove file_id to ensure only file_data is used file_element_file.pop("file_id", None) verbose_logger.debug( f"Converted file {file_id} from storage backend to base64 with format {content_type}" )