Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/container_endpoints/handler_factory.py: 59%
127 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""
2Factory for generating container proxy endpoints from JSON config.
4This module reads the endpoints.json config and dynamically creates
5FastAPI route handlers for ALL container file endpoints.
6"""
8import json
9from collections.abc import Mapping, Sequence
10from pathlib import Path
11from types import MappingProxyType
12from typing import Final
14from fastapi import APIRouter, Depends, Request, Response
15from fastapi.responses import ORJSONResponse
17from litellm.proxy._types import UserAPIKeyAuth
18from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
19from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
20from litellm.proxy.common_utils.openai_endpoint_utils import (
21 get_custom_llm_provider_from_request_headers,
22 get_custom_llm_provider_from_request_query,
23)
24from litellm.proxy.container_endpoints.ownership import (
25 assert_user_can_access_container,
26 get_container_forwarding_params,
27)
30def _load_endpoints_config() -> dict:
31 """Load the endpoints configuration from JSON file."""
32 config_path: Final = Path(__file__).parent.parent.parent / "containers" / "endpoints.json"
33 with open(config_path) as f:
34 return json.load(f)
37def get_all_route_types() -> list[str]:
38 """Get all async route types for registration in route_llm_request.py"""
39 config: Final = _load_endpoints_config()
40 return [endpoint["async_name"] for endpoint in config["endpoints"]]
43def _get_container_provider_config(custom_llm_provider: str):
44 """Get the container provider config for the given provider."""
45 if custom_llm_provider == "openai":
46 from litellm.llms.openai.containers.transformation import OpenAIContainerConfig
48 return OpenAIContainerConfig()
49 elif custom_llm_provider in ("azure", "azure_text"):
50 from litellm.llms.azure.containers.transformation import AzureContainerConfig
52 return AzureContainerConfig()
53 raise ValueError(f"Container API not supported for provider: {custom_llm_provider}")
56def _create_handler_for_path_params(
57 path_params: list[str],
58 route_type: str,
59 returns_binary: bool = False,
60 is_multipart: bool = False,
61 query_param_names: Sequence[str] = (),
62):
63 """
64 Dynamically create a handler with the correct path parameter signature.
65 """
66 # For binary content endpoints, use a different handler
67 if returns_binary and path_params == ["container_id", "file_id"]:
69 async def handler_binary_content(
70 request: Request,
71 container_id: str,
72 file_id: str,
73 fastapi_response: Response,
74 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
75 ):
76 return await _process_binary_request(
77 request=request,
78 fastapi_response=fastapi_response,
79 container_id=container_id,
80 file_id=file_id,
81 user_api_key_dict=user_api_key_dict,
82 )
84 return handler_binary_content
86 # For multipart file upload endpoints
87 if is_multipart:
89 async def handler_multipart_upload(
90 request: Request,
91 container_id: str,
92 fastapi_response: Response,
93 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
94 ):
95 return await _process_multipart_upload_request(
96 request=request,
97 fastapi_response=fastapi_response,
98 user_api_key_dict=user_api_key_dict,
99 route_type=route_type,
100 container_id=container_id,
101 )
103 return handler_multipart_upload
105 # Create handlers for different path parameter combinations
106 if path_params == ["container_id"]:
108 async def handler_container_id(
109 request: Request,
110 container_id: str,
111 fastapi_response: Response,
112 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
113 ):
114 return await _process_request(
115 request=request,
116 fastapi_response=fastapi_response,
117 user_api_key_dict=user_api_key_dict,
118 route_type=route_type,
119 path_params={"container_id": container_id},
120 query_param_names=query_param_names,
121 )
123 return handler_container_id
125 elif path_params == ["container_id", "file_id"]: 125 ↛ 147line 125 didn't jump to line 147 because the condition on line 125 was always true
127 async def handler_container_file(
128 request: Request,
129 container_id: str,
130 file_id: str,
131 fastapi_response: Response,
132 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
133 ):
134 return await _process_request(
135 request=request,
136 fastapi_response=fastapi_response,
137 user_api_key_dict=user_api_key_dict,
138 route_type=route_type,
139 path_params={"container_id": container_id, "file_id": file_id},
140 query_param_names=query_param_names,
141 )
143 return handler_container_file
145 else:
146 # Fallback for no path params
147 async def handler_no_params(
148 request: Request,
149 fastapi_response: Response,
150 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
151 ):
152 return await _process_request(
153 request=request,
154 fastapi_response=fastapi_response,
155 user_api_key_dict=user_api_key_dict,
156 route_type=route_type,
157 path_params={},
158 query_param_names=query_param_names,
159 )
161 return handler_no_params
164async def _process_binary_request(
165 request: Request,
166 fastapi_response: Response,
167 container_id: str,
168 file_id: str,
169 user_api_key_dict: UserAPIKeyAuth,
170):
171 """
172 Process binary content requests through the standard proxy/router pipeline.
174 Validate ownership before forwarding the provider-native container id through
175 the standard proxy/router pipeline. This handler only adapts the byte
176 response to FastAPI.
177 """
178 from litellm.proxy.proxy_server import (
179 general_settings,
180 llm_router,
181 proxy_config,
182 proxy_logging_obj,
183 select_data_generator,
184 user_api_base,
185 user_max_tokens,
186 user_model,
187 user_request_timeout,
188 user_temperature,
189 version,
190 )
192 custom_llm_provider: Final = (
193 get_custom_llm_provider_from_request_headers(request=request)
194 or get_custom_llm_provider_from_request_query(request=request)
195 or "openai"
196 )
198 original_container_id, resolved_provider = await assert_user_can_access_container(
199 container_id=container_id,
200 user_api_key_dict=user_api_key_dict,
201 custom_llm_provider=custom_llm_provider,
202 )
203 data: Final[dict[str, object]] = {
204 "file_id": file_id,
205 **(
206 await get_container_forwarding_params(
207 container_id=container_id,
208 original_container_id=original_container_id,
209 custom_llm_provider=resolved_provider,
210 )
211 ),
212 }
213 processor: Final = ProxyBaseLLMRequestProcessing(data=data)
215 try:
216 content: Final[object] = await processor.base_process_llm_request(
217 request=request,
218 fastapi_response=fastapi_response,
219 user_api_key_dict=user_api_key_dict,
220 route_type="aretrieve_container_file_content",
221 proxy_logging_obj=proxy_logging_obj,
222 llm_router=llm_router,
223 general_settings=general_settings,
224 proxy_config=proxy_config,
225 select_data_generator=select_data_generator,
226 model=None,
227 user_model=user_model,
228 user_temperature=user_temperature,
229 user_request_timeout=user_request_timeout,
230 user_max_tokens=user_max_tokens,
231 user_api_base=user_api_base,
232 version=version,
233 )
235 # Determine content type based on common file extensions in the file_id
236 content_type = "application/octet-stream"
237 file_id_lower: Final = file_id.lower()
238 if ".png" in file_id_lower or file_id_lower.endswith("png"):
239 content_type = "image/png"
240 elif ".jpg" in file_id_lower or ".jpeg" in file_id_lower:
241 content_type = "image/jpeg"
242 elif ".gif" in file_id_lower:
243 content_type = "image/gif"
244 elif ".csv" in file_id_lower:
245 content_type = "text/csv"
246 elif ".json" in file_id_lower:
247 content_type = "application/json"
248 elif ".txt" in file_id_lower:
249 content_type = "text/plain"
250 elif ".pdf" in file_id_lower:
251 content_type = "application/pdf"
253 if not isinstance(content, bytes):
254 raise TypeError(f"aretrieve_container_file_content expected bytes, got {type(content).__name__}")
256 return Response(
257 content=content,
258 headers=dict(fastapi_response.headers),
259 media_type=content_type,
260 )
262 except Exception as e:
263 raise await processor._handle_llm_api_exception(
264 e=e,
265 user_api_key_dict=user_api_key_dict,
266 proxy_logging_obj=proxy_logging_obj,
267 version=version,
268 )
271async def _process_multipart_upload_request(
272 request: Request,
273 fastapi_response: Response,
274 user_api_key_dict: UserAPIKeyAuth,
275 route_type: str,
276 container_id: str,
277) -> object:
278 """Process multipart file upload requests."""
279 from litellm.proxy.common_utils.http_parsing_utils import (
280 convert_upload_files_to_file_data,
281 get_form_data,
282 )
283 from litellm.proxy.proxy_server import (
284 general_settings,
285 llm_router,
286 proxy_config,
287 proxy_logging_obj,
288 select_data_generator,
289 user_api_base,
290 user_max_tokens,
291 user_model,
292 user_request_timeout,
293 user_temperature,
294 version,
295 )
297 # Parse multipart form data and convert files
298 form_data: Final = await get_form_data(request)
299 data: Final = await convert_upload_files_to_file_data(form_data)
301 if "file" not in data: 301 ↛ 307line 301 didn't jump to line 307 because the condition on line 301 was always true
302 from fastapi import HTTPException
304 raise HTTPException(status_code=400, detail="Missing required 'file' field")
306 # convert_upload_files_to_file_data returns list of tuples, extract single file
307 file_list: Final = data["file"]
308 if isinstance(file_list, list) and len(file_list) > 0:
309 data["file"] = file_list[0]
311 custom_llm_provider: Final = (
312 get_custom_llm_provider_from_request_headers(request=request)
313 or get_custom_llm_provider_from_request_query(request=request)
314 or "openai"
315 )
317 original_container_id, resolved_provider = await assert_user_can_access_container(
318 container_id=container_id,
319 user_api_key_dict=user_api_key_dict,
320 custom_llm_provider=custom_llm_provider,
321 )
323 data.update(
324 await get_container_forwarding_params(
325 container_id=container_id,
326 original_container_id=original_container_id,
327 custom_llm_provider=resolved_provider,
328 )
329 )
331 processor: Final = ProxyBaseLLMRequestProcessing(data=data)
332 try:
333 return await processor.base_process_llm_request(
334 request=request,
335 fastapi_response=fastapi_response,
336 user_api_key_dict=user_api_key_dict,
337 route_type=route_type,
338 proxy_logging_obj=proxy_logging_obj,
339 llm_router=llm_router,
340 general_settings=general_settings,
341 proxy_config=proxy_config,
342 select_data_generator=select_data_generator,
343 model=None,
344 user_model=user_model,
345 user_temperature=user_temperature,
346 user_request_timeout=user_request_timeout,
347 user_max_tokens=user_max_tokens,
348 user_api_base=user_api_base,
349 version=version,
350 )
351 except Exception as e:
352 raise await processor._handle_llm_api_exception(
353 e=e,
354 user_api_key_dict=user_api_key_dict,
355 proxy_logging_obj=proxy_logging_obj,
356 version=version,
357 )
360def _declared_query_params(query_params: Mapping[str, str], query_param_names: Sequence[str]) -> Mapping[str, str]:
361 return MappingProxyType({name: query_params[name] for name in query_param_names if name in query_params})
364async def _process_request(
365 request: Request,
366 fastapi_response: Response,
367 user_api_key_dict: UserAPIKeyAuth,
368 route_type: str,
369 path_params: dict[str, str],
370 query_param_names: Sequence[str] = (),
371) -> object:
372 """Common request processing logic."""
373 from litellm.proxy.proxy_server import (
374 general_settings,
375 llm_router,
376 proxy_config,
377 proxy_logging_obj,
378 select_data_generator,
379 user_api_base,
380 user_max_tokens,
381 user_model,
382 user_request_timeout,
383 user_temperature,
384 version,
385 )
387 query_params: Final = dict(request.query_params)
388 data: Final[dict[str, object]] = {
389 "query_params": query_params,
390 **_declared_query_params(query_params, query_param_names),
391 **path_params,
392 }
394 custom_llm_provider: Final = (
395 get_custom_llm_provider_from_request_headers(request=request)
396 or get_custom_llm_provider_from_request_query(request=request)
397 or "openai"
398 )
400 # Validate container_id ownership if present in path_params.
401 if "container_id" in path_params: 401 ↛ 418line 401 didn't jump to line 418 because the condition on line 401 was always true
402 (
403 original_container_id,
404 resolved_provider,
405 ) = await assert_user_can_access_container(
406 container_id=path_params["container_id"],
407 user_api_key_dict=user_api_key_dict,
408 custom_llm_provider=custom_llm_provider,
409 )
410 data.update(
411 await get_container_forwarding_params(
412 container_id=path_params["container_id"],
413 original_container_id=original_container_id,
414 custom_llm_provider=resolved_provider,
415 )
416 )
417 else:
418 data["custom_llm_provider"] = custom_llm_provider
420 processor: Final = ProxyBaseLLMRequestProcessing(data=data)
421 try:
422 return await processor.base_process_llm_request(
423 request=request,
424 fastapi_response=fastapi_response,
425 user_api_key_dict=user_api_key_dict,
426 route_type=route_type,
427 proxy_logging_obj=proxy_logging_obj,
428 llm_router=llm_router,
429 general_settings=general_settings,
430 proxy_config=proxy_config,
431 select_data_generator=select_data_generator,
432 model=None,
433 user_model=user_model,
434 user_temperature=user_temperature,
435 user_request_timeout=user_request_timeout,
436 user_max_tokens=user_max_tokens,
437 user_api_base=user_api_base,
438 version=version,
439 )
440 except Exception as e:
441 raise await processor._handle_llm_api_exception(
442 e=e,
443 user_api_key_dict=user_api_key_dict,
444 proxy_logging_obj=proxy_logging_obj,
445 version=version,
446 )
449def register_container_file_endpoints(router: APIRouter) -> None:
450 """
451 Register ALL container file endpoints from JSON config to the router.
453 This single function registers all endpoints defined in endpoints.json,
454 eliminating the need for manual endpoint definitions.
455 """
456 config: Final = _load_endpoints_config()
458 for endpoint_config in config["endpoints"]:
459 path = endpoint_config["path"]
460 method = endpoint_config["method"].lower()
461 path_params = endpoint_config.get("path_params", [])
462 route_type = endpoint_config["async_name"]
463 returns_binary = endpoint_config.get("returns_binary", False)
464 is_multipart = endpoint_config.get("is_multipart", False)
466 # Create handler with correct signature for path params
467 handler = _create_handler_for_path_params(
468 path_params,
469 route_type,
470 returns_binary,
471 is_multipart,
472 query_param_names=endpoint_config.get("query_params", ()),
473 )
475 # Register routes
476 route_method = getattr(router, method)
478 # For binary endpoints, don't use ORJSONResponse
479 if returns_binary:
480 # Register both /v1/... and /... paths without JSON response class
481 route_method(
482 f"/v1{path}",
483 dependencies=[Depends(user_api_key_auth)],
484 tags=["containers"],
485 )(handler)
487 route_method(
488 path,
489 dependencies=[Depends(user_api_key_auth)],
490 tags=["containers"],
491 )(handler)
492 else:
493 # Register both /v1/... and /... paths with JSON response
494 route_method(
495 f"/v1{path}",
496 dependencies=[Depends(user_api_key_auth)],
497 response_class=ORJSONResponse,
498 tags=["containers"],
499 )(handler)
501 route_method(
502 path,
503 dependencies=[Depends(user_api_key_auth)],
504 response_class=ORJSONResponse,
505 tags=["containers"],
506 )(handler)