Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/credential_endpoints/endpoints.py: 57%
160 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""
2CRUD endpoints for storing reusable credentials.
3"""
5from collections.abc import Mapping
6from typing import (
7 Annotated,
8 Final,
9 cast, # noqa: TID251 # jsonify_object in proxy/utils.py is annotated with a bare dict
10)
12from fastapi import APIRouter, Depends, HTTPException, Path, Request, Response
13from pydantic import TypeAdapter
15import litellm
16from litellm._logging import verbose_proxy_logger
17from litellm.litellm_core_utils.credential_accessor import CredentialAccessor
18from litellm.litellm_core_utils.litellm_logging import _get_masked_values
19from litellm.models.credentials import UpdateCredentialItem
20from litellm.proxy._types import CommonProxyErrors, UserAPIKeyAuth
21from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
22from litellm.proxy.common_utils.encrypt_decrypt_utils import encrypt_value_helper
23from litellm.proxy.utils import handle_exception_on_proxy, jsonify_object
24from litellm.repositories.base_repository import is_unique_violation
25from litellm.repositories.credentials_repository import CredentialsRepository
26from litellm.types.utils import CreateCredentialItem, CredentialItem
28router: Final = APIRouter()
29_CREDENTIAL_DICT_ADAPTER: Final = TypeAdapter(dict[str, object])
32class CredentialHelperUtils:
33 @staticmethod
34 def encrypt_credential_values(credential: CredentialItem, new_encryption_key: str | None = None) -> CredentialItem:
35 """Encrypt values in credential.credential_values and add to DB"""
36 encrypted_credential_values: Final = {}
37 for key, value in (credential.credential_values or {}).items():
38 encrypted_credential_values[key] = encrypt_value_helper(value, new_encryption_key)
40 # Return a new object to avoid mutating the caller's credential, which
41 # is kept in memory and should remain unencrypted.
42 return CredentialItem(
43 credential_name=credential.credential_name,
44 credential_values=encrypted_credential_values,
45 credential_info=credential.credential_info or {},
46 )
49def _credential_exists_detail(credential_name: str) -> str:
50 return (
51 f"Credential '{credential_name}' already exists. "
52 f"Update it with PATCH /credentials/{credential_name}, or delete it first."
53 )
56def get_llm_router() -> litellm.Router | None:
57 from litellm.proxy.proxy_server import llm_router
59 return llm_router
62def _resolve_deployment_credentials(llm_router: litellm.Router | None, model_id: str) -> Mapping[str, object]:
63 if llm_router is None: 63 ↛ 64line 63 didn't jump to line 64 because the condition on line 63 was never true
64 raise HTTPException(
65 status_code=500,
66 detail="LLM router not found. Please ensure you have a valid router instance.",
67 )
68 if llm_router.get_deployment(model_id) is None: 68 ↛ 70line 68 didn't jump to line 70 because the condition on line 68 was always true
69 raise HTTPException(status_code=404, detail="Model not found")
70 credential_values: Final = llm_router.get_deployment_credentials(model_id)
71 if credential_values is None:
72 raise HTTPException(status_code=404, detail="Model not found")
73 return _CREDENTIAL_DICT_ADAPTER.validate_python(credential_values)
76@router.post(
77 "/credentials",
78 dependencies=[Depends(user_api_key_auth)],
79 tags=["credential management"],
80)
81async def create_credential(
82 request: Request,
83 fastapi_response: Response,
84 credential: CreateCredentialItem,
85 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
86 llm_router: Annotated[litellm.Router | None, Depends(get_llm_router)] = None,
87):
88 """
89 [BETA] endpoint. This might change unexpectedly.
90 Stores credential in DB.
91 Reloads credentials in memory.
92 """
93 from litellm.proxy.proxy_server import prisma_client
95 try:
96 if prisma_client is None: 96 ↛ 97line 96 didn't jump to line 97 because the condition on line 96 was never true
97 raise HTTPException(
98 status_code=500,
99 detail={"error": CommonProxyErrors.db_not_connected_error.value},
100 )
101 credential_values: Final = (
102 _resolve_deployment_credentials(llm_router, credential.model_id)
103 if credential.model_id
104 else credential.credential_values
105 )
106 if credential_values is None: 106 ↛ 107line 106 didn't jump to line 107 because the condition on line 106 was never true
107 raise HTTPException(
108 status_code=400,
109 detail="Credential values are required. Unable to infer credential values from model ID.",
110 )
111 processed_credential: Final = CredentialItem(
112 credential_name=credential.credential_name,
113 credential_values=_CREDENTIAL_DICT_ADAPTER.validate_python(credential_values),
114 credential_info=credential.credential_info,
115 )
116 encrypted_credential: Final = CredentialHelperUtils.encrypt_credential_values(processed_credential)
117 credentials_dict: Final = encrypted_credential.model_dump()
118 credentials_dict_jsonified: Final = cast( # cast-ok: deep-copies a model_dump, so keys are str
119 "dict[str, object]", jsonify_object(credentials_dict)
120 )
121 try:
122 await CredentialsRepository(prisma_client).create(
123 data={
124 **credentials_dict_jsonified,
125 "created_by": user_api_key_dict.user_id,
126 "updated_by": user_api_key_dict.user_id,
127 }
128 )
129 except Exception as e:
130 if not is_unique_violation(e):
131 raise
132 raise HTTPException(status_code=409, detail=_credential_exists_detail(credential.credential_name))
134 ## ADD TO LITELLM ##
135 CredentialAccessor.upsert_credentials([processed_credential])
137 return {"success": True, "message": "Credential created successfully"}
138 except Exception as e:
139 verbose_proxy_logger.exception(e)
140 raise handle_exception_on_proxy(e)
143@router.get(
144 "/credentials",
145 dependencies=[Depends(user_api_key_auth)],
146 tags=["credential management"],
147)
148async def get_credentials(
149 request: Request,
150 fastapi_response: Response,
151 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
152):
153 """
154 [BETA] endpoint. This might change unexpectedly.
155 """
156 try:
157 masked_credentials: Final = [
158 {
159 "credential_name": credential.credential_name,
160 "credential_values": _get_masked_values(credential.credential_values),
161 "credential_info": credential.credential_info,
162 }
163 for credential in litellm.credential_list
164 ]
165 return {"success": True, "credentials": masked_credentials}
166 except Exception as e:
167 raise handle_exception_on_proxy(e)
170@router.get(
171 "/credentials/by_name/{credential_name:path}",
172 dependencies=[Depends(user_api_key_auth)],
173 tags=["credential management"],
174 response_model=CredentialItem,
175)
176async def get_credential_by_name(
177 request: Request,
178 fastapi_response: Response,
179 credential_name: str = Path(..., description="The credential name, percent-decoded; may contain slashes"),
180 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
181):
182 """
183 [BETA] endpoint. This might change unexpectedly.
184 """
185 try:
186 for credential in litellm.credential_list:
187 if credential.credential_name == credential_name: 187 ↛ 188line 187 didn't jump to line 188 because the condition on line 187 was never true
188 masked_credential = CredentialItem(
189 credential_name=credential.credential_name,
190 credential_values=_get_masked_values(
191 credential.credential_values,
192 unmasked_length=4,
193 number_of_asterisks=4,
194 ),
195 credential_info=credential.credential_info,
196 )
197 return masked_credential
198 raise HTTPException(
199 status_code=404,
200 detail="Credential not found. Got credential name: " + credential_name,
201 )
202 except Exception as e:
203 verbose_proxy_logger.exception(e)
204 raise handle_exception_on_proxy(e)
207@router.get(
208 "/credentials/by_model/{model_id}",
209 dependencies=[Depends(user_api_key_auth)],
210 tags=["credential management"],
211 response_model=CredentialItem,
212)
213async def get_credential_by_model(
214 request: Request,
215 fastapi_response: Response,
216 model_id: str = Path(..., description="The model ID to look up credentials for"),
217 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
218):
219 """
220 [BETA] endpoint. This might change unexpectedly.
221 """
222 from litellm.proxy.proxy_server import llm_router
224 try:
225 if llm_router is None: 225 ↛ 226line 225 didn't jump to line 226 because the condition on line 225 was never true
226 raise HTTPException(status_code=500, detail="LLM router not found")
227 model: Final = llm_router.get_deployment(model_id)
228 if model is None: 228 ↛ 230line 228 didn't jump to line 230 because the condition on line 228 was always true
229 raise HTTPException(status_code=404, detail="Model not found")
230 credential_values: Final = llm_router.get_deployment_credentials(model_id)
231 if credential_values is None:
232 raise HTTPException(status_code=404, detail="Model not found")
233 masked_credential_values: Final = _get_masked_values(
234 credential_values,
235 unmasked_length=4,
236 number_of_asterisks=4,
237 )
238 credential: Final = CredentialItem(
239 credential_name=f"{model.model_name}-credential-{model_id}",
240 credential_values=masked_credential_values,
241 credential_info={},
242 )
243 return credential
244 except Exception as e:
245 verbose_proxy_logger.exception(e)
246 raise handle_exception_on_proxy(e)
249@router.delete(
250 "/credentials/{credential_name:path}",
251 dependencies=[Depends(user_api_key_auth)],
252 tags=["credential management"],
253)
254async def delete_credential(
255 request: Request,
256 fastapi_response: Response,
257 credential_name: str = Path(..., description="The credential name, percent-decoded; may contain slashes"),
258 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
259):
260 """
261 [BETA] endpoint. This might change unexpectedly.
262 """
263 from litellm.proxy.proxy_server import prisma_client
265 try:
266 if prisma_client is None: 266 ↛ 267line 266 didn't jump to line 267 because the condition on line 266 was never true
267 raise HTTPException(
268 status_code=500,
269 detail={"error": CommonProxyErrors.db_not_connected_error.value},
270 )
271 deleted: Final = await CredentialsRepository(prisma_client).delete_by_name(credential_name)
272 if deleted is None: 272 ↛ 279line 272 didn't jump to line 279 because the condition on line 272 was always true
273 raise HTTPException(
274 status_code=404,
275 detail="Credential not found. Got credential name: " + credential_name,
276 )
278 ## DELETE FROM LITELLM ##
279 litellm.credential_list = [cred for cred in litellm.credential_list if cred.credential_name != credential_name]
280 return {"success": True, "message": "Credential deleted successfully"}
281 except Exception as e:
282 raise handle_exception_on_proxy(e)
285def update_db_credential(
286 db_credential: CredentialItem,
287 updated_patch: CredentialItem,
288 new_encryption_key: str | None = None,
289) -> CredentialItem:
290 """
291 Update a credential in the DB.
292 """
293 merged_credential: Final = CredentialItem(
294 credential_name=db_credential.credential_name,
295 credential_info=db_credential.credential_info,
296 credential_values=db_credential.credential_values,
297 )
299 encrypted_credential: Final = CredentialHelperUtils.encrypt_credential_values(
300 updated_patch,
301 new_encryption_key,
302 )
303 # update model name
304 if encrypted_credential.credential_name:
305 merged_credential.credential_name = encrypted_credential.credential_name
307 # update litellm params
308 if encrypted_credential.credential_values:
309 # Encrypt any sensitive values
310 encrypted_params: Final = {k: v for k, v in encrypted_credential.credential_values.items()}
312 merged_credential.credential_values.update(encrypted_params)
314 # update model info
315 if encrypted_credential.credential_info:
316 """Update credential info"""
317 if "credential_info" not in merged_credential.credential_info:
318 merged_credential.credential_info = {}
319 merged_credential.credential_info.update(encrypted_credential.credential_info)
321 return merged_credential
324@router.patch(
325 "/credentials/{credential_name:path}",
326 dependencies=[Depends(user_api_key_auth)],
327 tags=["credential management"],
328)
329async def update_credential(
330 request: Request,
331 fastapi_response: Response,
332 credential: UpdateCredentialItem,
333 credential_name: str = Path(..., description="The credential name, percent-decoded; may contain slashes"),
334 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
335 llm_router: Annotated[litellm.Router | None, Depends(get_llm_router)] = None,
336):
337 """
338 [BETA] endpoint. This might change unexpectedly.
339 """
340 from litellm.proxy.proxy_server import prisma_client
342 try:
343 if prisma_client is None: 343 ↛ 344line 343 didn't jump to line 344 because the condition on line 343 was never true
344 raise HTTPException(
345 status_code=500,
346 detail={"error": CommonProxyErrors.db_not_connected_error.value},
347 )
348 credentials_repository: Final = CredentialsRepository(prisma_client)
349 db_credential: Final = await credentials_repository.find_by_name(credential_name)
350 if db_credential is None: 350 ↛ 352line 350 didn't jump to line 352 because the condition on line 350 was always true
351 raise HTTPException(status_code=404, detail="Credential not found in DB.")
352 patch: Final = CredentialItem(
353 credential_name=credential.credential_name,
354 credential_info=_CREDENTIAL_DICT_ADAPTER.validate_python(credential.credential_info),
355 credential_values=_CREDENTIAL_DICT_ADAPTER.validate_python(
356 _resolve_deployment_credentials(llm_router, credential.model_id)
357 if credential.model_id
358 else credential.credential_values or {}
359 ),
360 )
361 merged_credential: Final = update_db_credential(db_credential, patch)
362 credential_object_jsonified: Final = cast( # cast-ok: deep-copies a model_dump, so keys are str
363 "dict[str, object]", jsonify_object(merged_credential.model_dump())
364 )
365 await credentials_repository.update_by_name(
366 credential_name,
367 data={
368 **credential_object_jsonified,
369 "updated_by": user_api_key_dict.user_id,
370 },
371 )
373 # Sync in-memory credential_list (skip if not in memory - e.g., proxy restarted)
374 new_name: Final = merged_credential.credential_name
375 existing_in_memory: CredentialItem | None = None
376 for cred in litellm.credential_list:
377 if cred.credential_name == credential_name:
378 existing_in_memory = cred
379 break
381 if existing_in_memory is not None:
382 in_memory_values: Final = dict(existing_in_memory.credential_values or {})
383 if patch.credential_values:
384 in_memory_values.update(patch.credential_values)
385 in_memory_info: Final = dict(existing_in_memory.credential_info or {})
386 if patch.credential_info:
387 in_memory_info.update(patch.credential_info)
388 updated_in_memory: Final = CredentialItem(
389 credential_name=new_name,
390 credential_values=in_memory_values,
391 credential_info=in_memory_info,
392 )
393 # Remove old entry if renamed, then use upsert_credentials to handle duplicates
394 if new_name != credential_name:
395 litellm.credential_list = [c for c in litellm.credential_list if c.credential_name != credential_name]
396 CredentialAccessor.upsert_credentials([updated_in_memory])
398 return {"success": True, "message": "Credential updated successfully"}
399 except Exception as e:
400 raise handle_exception_on_proxy(e)