Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/openai_files_endpoints/storage_backend_service.py: 30%
70 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""
2Storage backend service for file upload operations.
4This module provides a service class for handling file uploads to custom
5storage backends (e.g., Azure Blob Storage) and managing associated metadata.
6"""
8import base64
9import time
10from collections.abc import Sequence
11from typing import Any, Final, cast
13from litellm._logging import verbose_proxy_logger
14from litellm._uuid import uuid as uuid_module
15from litellm.llms.base_llm.files.storage_backend import BaseFileStorageBackend
16from litellm.llms.base_llm.files.storage_backend_factory import get_storage_backend
17from litellm.llms.base_llm.files.transformation import BaseFileEndpoints
18from litellm.proxy._types import ProxyException, UserAPIKeyAuth
19from litellm.proxy.utils import PrismaClient, ProxyLogging
20from litellm.types.llms.openai import OpenAIFileObject, OpenAIFilesPurpose
21from litellm.types.utils import ExtractedFileData, SpecialEnums
24class StorageBackendFileService:
25 """
26 Service for handling file uploads to storage backends.
28 This service encapsulates the logic for:
29 - Uploading files to storage backends
30 - Creating file objects with storage metadata
31 - Generating unified file IDs for managed files
32 - Storing files in the managed files system
33 """
35 @staticmethod
36 async def upload_file_to_storage_backend(
37 file_data: ExtractedFileData,
38 target_storage: str,
39 target_model_names: Sequence[str],
40 purpose: OpenAIFilesPurpose,
41 proxy_logging_obj: ProxyLogging,
42 user_api_key_dict: UserAPIKeyAuth,
43 prisma_client: PrismaClient | None = None,
44 ) -> OpenAIFileObject:
45 """
46 Upload a file to a storage backend and create a file object.
48 Args:
49 file_data: File data dictionary from extract_file_data()
50 target_storage: Storage backend name (e.g., "azure_storage", "litellm_db")
51 target_model_names: List of model names for managed files
52 purpose: File purpose (e.g., "user_data", "batch")
53 proxy_logging_obj: Proxy logging object for accessing hooks
54 user_api_key_dict: User API key authentication data
55 prisma_client: The proxy's database client, required by the "litellm_db" backend
57 Returns:
58 OpenAIFileObject: Created file object with storage metadata
60 Raises:
61 ProxyException: If storage backend is invalid or upload fails
62 """
63 # Get storage backend instance
64 try:
65 storage_backend: Final = get_storage_backend(target_storage, prisma_client=prisma_client)
66 except ValueError as e:
67 raise ProxyException(
68 message=str(e),
69 type="invalid_request_error",
70 param="target_storage",
71 code=400,
72 )
74 if target_model_names:
75 managed_files_hook: Final = proxy_logging_obj.get_proxy_hook("managed_files")
76 if not isinstance(managed_files_hook, BaseFileEndpoints):
77 raise ProxyException(
78 message="Uploading with target_model_names requires a database-connected proxy, and this proxy has no database configured",
79 type="invalid_request_error",
80 param="target_model_names",
81 code=400,
82 )
84 # Extract file information
85 file_content: Final = file_data["content"]
86 filename: Final = file_data.get("filename", "file")
87 content_type: Final = file_data.get("content_type", "application/octet-stream")
89 # Upload to storage backend
90 storage_url: Final = await storage_backend.upload_file(
91 file_content=file_content,
92 filename=filename,
93 content_type=content_type,
94 path_prefix="",
95 file_naming_strategy="uuid",
96 )
98 verbose_proxy_logger.debug("Storage backend upload complete: backend=%s, url=%s", target_storage, storage_url)
100 # Create file object with storage metadata
101 file_object: Final = StorageBackendFileService._create_file_object_with_storage_metadata(
102 file_content=file_content,
103 filename=filename,
104 purpose=purpose,
105 target_storage=target_storage,
106 storage_url=storage_url,
107 )
109 if not target_model_names:
110 return file_object
111 try:
112 await StorageBackendFileService._store_in_managed_files(
113 file_object=file_object,
114 file_data=file_data,
115 target_model_names=target_model_names,
116 target_storage=target_storage,
117 storage_url=storage_url,
118 proxy_logging_obj=proxy_logging_obj,
119 user_api_key_dict=user_api_key_dict,
120 )
121 except Exception:
122 await StorageBackendFileService._discard_orphaned_content(storage_backend, storage_url, target_storage)
123 raise
124 return file_object
126 @staticmethod
127 async def _discard_orphaned_content(
128 storage_backend: BaseFileStorageBackend, storage_url: str, target_storage: str
129 ) -> None:
130 try:
131 await storage_backend.delete_file(storage_url)
132 except Exception as e: # noqa: BLE001 # the metadata failure is what surfaces; a failed cleanup is only logged
133 verbose_proxy_logger.warning(
134 "Could not delete orphaned content at %s on %s after its metadata write failed: %s",
135 storage_url,
136 target_storage,
137 e,
138 )
140 @staticmethod
141 def _create_file_object_with_storage_metadata(
142 file_content: bytes,
143 filename: str,
144 purpose: OpenAIFilesPurpose,
145 target_storage: str,
146 storage_url: str,
147 ) -> OpenAIFileObject:
148 """
149 Create an OpenAIFileObject with storage backend metadata.
151 Args:
152 file_content: File content bytes
153 filename: Original filename
154 purpose: File purpose
155 target_storage: Storage backend name
156 storage_url: URL where file is stored
158 Returns:
159 OpenAIFileObject: File object with storage metadata in _hidden_params
160 """
161 file_id: Final = f"file-{uuid_module.uuid4().hex[:24]}"
162 file_object: Final = OpenAIFileObject(
163 id=file_id,
164 object="file",
165 purpose=purpose,
166 created_at=int(time.time()),
167 bytes=len(file_content),
168 filename=filename,
169 status="uploaded",
170 )
172 # Store storage metadata in hidden params
173 if not hasattr(file_object, "_hidden_params") or file_object._hidden_params is None:
174 file_object._hidden_params = {}
175 file_object._hidden_params.update(
176 {
177 "storage_backend": target_storage,
178 "storage_url": storage_url,
179 }
180 )
182 return file_object
184 @staticmethod
185 def _create_unified_file_id(
186 file_type: str | None,
187 target_model_names: Sequence[str],
188 file_id: str,
189 ) -> str:
190 """
191 Create a base64-encoded unified file ID for managed files.
193 Args:
194 file_type: MIME type of the file
195 target_model_names: List of model names
196 file_id: Original file ID
198 Returns:
199 str: Base64-encoded unified file ID
200 """
201 unified_file_id_str: Final = SpecialEnums.LITELLM_MANAGED_FILE_COMPLETE_STR.value.format(
202 file_type,
203 str(uuid_module.uuid4()),
204 ",".join(target_model_names),
205 file_id,
206 None,
207 )
209 base64_unified_file_id: Final = base64.urlsafe_b64encode(unified_file_id_str.encode()).decode().rstrip("=")
211 return base64_unified_file_id
213 @staticmethod
214 async def _store_in_managed_files(
215 file_object: OpenAIFileObject,
216 file_data: ExtractedFileData,
217 target_model_names: Sequence[str],
218 target_storage: str,
219 storage_url: str,
220 proxy_logging_obj: ProxyLogging,
221 user_api_key_dict: UserAPIKeyAuth,
222 ) -> None:
223 """
224 Store file in managed files system with unified file ID.
226 Args:
227 file_object: File object to store
228 file_data: File data dictionary
229 target_model_names: List of model names
230 target_storage: Storage backend name
231 storage_url: URL where file is stored
232 proxy_logging_obj: Proxy logging object
233 user_api_key_dict: User API key authentication data
234 """
235 managed_files_obj = proxy_logging_obj.get_proxy_hook("managed_files")
236 if not managed_files_obj or not isinstance(managed_files_obj, BaseFileEndpoints):
237 verbose_proxy_logger.warning("Managed files hook not available, skipping managed files storage")
238 return
239 managed_files_obj = cast(Any, managed_files_obj)
241 # Create model mappings using storage URL
242 model_mappings: Final = {model_name: storage_url for model_name in target_model_names}
244 # Create unified file ID
245 file_type: Final = file_data.get("content_type", "application/octet-stream")
246 base64_unified_file_id: Final = StorageBackendFileService._create_unified_file_id(
247 file_type=file_type,
248 target_model_names=target_model_names,
249 file_id=file_object.id,
250 )
252 # Update file object ID to unified ID
253 file_object.id = base64_unified_file_id
255 verbose_proxy_logger.debug(
256 "Storing file in managed files: unified_id=%s, storage_backend=%s, storage_url=%s",
257 base64_unified_file_id,
258 target_storage,
259 storage_url,
260 )
262 # Store in managed files
263 await managed_files_obj.store_unified_file_id(
264 file_id=base64_unified_file_id,
265 file_object=file_object,
266 litellm_parent_otel_span=user_api_key_dict.parent_otel_span,
267 model_mappings=model_mappings,
268 user_api_key_dict=user_api_key_dict,
269 )