mirror of
https://github.com/BerriAI/litellm.git
synced 2025-04-25 10:44:24 +00:00
319 lines
11 KiB
Python
319 lines
11 KiB
Python
"""
|
|
CRUD endpoints for storing reusable credentials.
|
|
"""
|
|
|
|
from typing import Optional, Union
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, Response
|
|
|
|
import litellm
|
|
from litellm._logging import verbose_proxy_logger
|
|
from litellm.litellm_core_utils.credential_accessor import CredentialAccessor
|
|
from litellm.litellm_core_utils.litellm_logging import _get_masked_values
|
|
from litellm.proxy._types import CommonProxyErrors, UserAPIKeyAuth
|
|
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
|
|
from litellm.proxy.common_utils.encrypt_decrypt_utils import encrypt_value_helper
|
|
from litellm.proxy.utils import handle_exception_on_proxy, jsonify_object
|
|
from litellm.types.utils import CreateCredentialItem, CredentialItem
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
class CredentialHelperUtils:
|
|
@staticmethod
|
|
def encrypt_credential_values(credential: CredentialItem) -> CredentialItem:
|
|
"""Encrypt values in credential.credential_values and add to DB"""
|
|
encrypted_credential_values = {}
|
|
for key, value in credential.credential_values.items():
|
|
encrypted_credential_values[key] = encrypt_value_helper(value)
|
|
credential.credential_values = encrypted_credential_values
|
|
return credential
|
|
|
|
|
|
@router.post(
|
|
"/credentials",
|
|
dependencies=[Depends(user_api_key_auth)],
|
|
tags=["credential management"],
|
|
)
|
|
async def create_credential(
|
|
request: Request,
|
|
fastapi_response: Response,
|
|
credential: CreateCredentialItem,
|
|
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
|
|
):
|
|
"""
|
|
[BETA] endpoint. This might change unexpectedly.
|
|
Stores credential in DB.
|
|
Reloads credentials in memory.
|
|
"""
|
|
from litellm.proxy.proxy_server import llm_router, prisma_client
|
|
|
|
try:
|
|
if prisma_client is None:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail={"error": CommonProxyErrors.db_not_connected_error.value},
|
|
)
|
|
if credential.model_id:
|
|
if llm_router is None:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail="LLM router not found. Please ensure you have a valid router instance.",
|
|
)
|
|
# get model from router
|
|
model = llm_router.get_deployment(credential.model_id)
|
|
if model is None:
|
|
raise HTTPException(status_code=404, detail="Model not found")
|
|
credential_values = llm_router.get_deployment_credentials(
|
|
credential.model_id
|
|
)
|
|
if credential_values is None:
|
|
raise HTTPException(status_code=404, detail="Model not found")
|
|
credential.credential_values = credential_values
|
|
|
|
if credential.credential_values is None:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Credential values are required. Unable to infer credential values from model ID.",
|
|
)
|
|
processed_credential = CredentialItem(
|
|
credential_name=credential.credential_name,
|
|
credential_values=credential.credential_values,
|
|
credential_info=credential.credential_info,
|
|
)
|
|
encrypted_credential = CredentialHelperUtils.encrypt_credential_values(
|
|
processed_credential
|
|
)
|
|
credentials_dict = encrypted_credential.model_dump()
|
|
credentials_dict_jsonified = jsonify_object(credentials_dict)
|
|
await prisma_client.db.litellm_credentialstable.create(
|
|
data={
|
|
**credentials_dict_jsonified,
|
|
"created_by": user_api_key_dict.user_id,
|
|
"updated_by": user_api_key_dict.user_id,
|
|
}
|
|
)
|
|
|
|
## ADD TO LITELLM ##
|
|
CredentialAccessor.upsert_credentials([processed_credential])
|
|
|
|
return {"success": True, "message": "Credential created successfully"}
|
|
except Exception as e:
|
|
verbose_proxy_logger.exception(e)
|
|
raise handle_exception_on_proxy(e)
|
|
|
|
|
|
@router.get(
|
|
"/credentials",
|
|
dependencies=[Depends(user_api_key_auth)],
|
|
tags=["credential management"],
|
|
)
|
|
async def get_credentials(
|
|
request: Request,
|
|
fastapi_response: Response,
|
|
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
|
|
):
|
|
"""
|
|
[BETA] endpoint. This might change unexpectedly.
|
|
"""
|
|
try:
|
|
masked_credentials = [
|
|
{
|
|
"credential_name": credential.credential_name,
|
|
"credential_values": _get_masked_values(credential.credential_values),
|
|
"credential_info": credential.credential_info,
|
|
}
|
|
for credential in litellm.credential_list
|
|
]
|
|
return {"success": True, "credentials": masked_credentials}
|
|
except Exception as e:
|
|
return handle_exception_on_proxy(e)
|
|
|
|
|
|
@router.get(
|
|
"/credentials/by_name/{credential_name}",
|
|
dependencies=[Depends(user_api_key_auth)],
|
|
tags=["credential management"],
|
|
response_model=CredentialItem,
|
|
)
|
|
@router.get(
|
|
"/credentials/by_model/{model_id}",
|
|
dependencies=[Depends(user_api_key_auth)],
|
|
tags=["credential management"],
|
|
response_model=CredentialItem,
|
|
)
|
|
async def get_credential(
|
|
request: Request,
|
|
fastapi_response: Response,
|
|
credential_name: Optional[str] = None,
|
|
model_id: Optional[str] = None,
|
|
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
|
|
):
|
|
"""
|
|
[BETA] endpoint. This might change unexpectedly.
|
|
"""
|
|
from litellm.proxy.proxy_server import llm_router
|
|
|
|
try:
|
|
if model_id:
|
|
if llm_router is None:
|
|
raise HTTPException(status_code=500, detail="LLM router not found")
|
|
model = llm_router.get_deployment(model_id)
|
|
if model is None:
|
|
raise HTTPException(status_code=404, detail="Model not found")
|
|
credential_values = llm_router.get_deployment_credentials(model_id)
|
|
if credential_values is None:
|
|
raise HTTPException(status_code=404, detail="Model not found")
|
|
masked_credential_values = _get_masked_values(
|
|
credential_values,
|
|
unmasked_length=4,
|
|
number_of_asterisks=4,
|
|
)
|
|
credential = CredentialItem(
|
|
credential_name="{}-credential-{}".format(model.model_name, model_id),
|
|
credential_values=masked_credential_values,
|
|
credential_info={},
|
|
)
|
|
# return credential object
|
|
return credential
|
|
elif credential_name:
|
|
for credential in litellm.credential_list:
|
|
if credential.credential_name == credential_name:
|
|
masked_credential = CredentialItem(
|
|
credential_name=credential.credential_name,
|
|
credential_values=_get_masked_values(
|
|
credential.credential_values,
|
|
unmasked_length=4,
|
|
number_of_asterisks=4,
|
|
),
|
|
credential_info=credential.credential_info,
|
|
)
|
|
return masked_credential
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="Credential not found. Got credential name: " + credential_name,
|
|
)
|
|
else:
|
|
raise HTTPException(
|
|
status_code=404, detail="Credential name or model ID required"
|
|
)
|
|
except Exception as e:
|
|
verbose_proxy_logger.exception(e)
|
|
raise handle_exception_on_proxy(e)
|
|
|
|
|
|
@router.delete(
|
|
"/credentials/{credential_name}",
|
|
dependencies=[Depends(user_api_key_auth)],
|
|
tags=["credential management"],
|
|
)
|
|
async def delete_credential(
|
|
request: Request,
|
|
fastapi_response: Response,
|
|
credential_name: str,
|
|
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
|
|
):
|
|
"""
|
|
[BETA] endpoint. This might change unexpectedly.
|
|
"""
|
|
from litellm.proxy.proxy_server import prisma_client
|
|
|
|
try:
|
|
if prisma_client is None:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail={"error": CommonProxyErrors.db_not_connected_error.value},
|
|
)
|
|
await prisma_client.db.litellm_credentialstable.delete(
|
|
where={"credential_name": credential_name}
|
|
)
|
|
|
|
## DELETE FROM LITELLM ##
|
|
litellm.credential_list = [
|
|
cred
|
|
for cred in litellm.credential_list
|
|
if cred.credential_name != credential_name
|
|
]
|
|
return {"success": True, "message": "Credential deleted successfully"}
|
|
except Exception as e:
|
|
return handle_exception_on_proxy(e)
|
|
|
|
|
|
def update_db_credential(
|
|
db_credential: CredentialItem, updated_patch: CredentialItem
|
|
) -> CredentialItem:
|
|
"""
|
|
Update a credential in the DB.
|
|
"""
|
|
merged_credential = CredentialItem(
|
|
credential_name=db_credential.credential_name,
|
|
credential_info=db_credential.credential_info,
|
|
credential_values=db_credential.credential_values,
|
|
)
|
|
|
|
encrypted_credential = CredentialHelperUtils.encrypt_credential_values(
|
|
updated_patch
|
|
)
|
|
# update model name
|
|
if encrypted_credential.credential_name:
|
|
merged_credential.credential_name = encrypted_credential.credential_name
|
|
|
|
# update litellm params
|
|
if encrypted_credential.credential_values:
|
|
# Encrypt any sensitive values
|
|
encrypted_params = {
|
|
k: v for k, v in encrypted_credential.credential_values.items()
|
|
}
|
|
|
|
merged_credential.credential_values.update(encrypted_params)
|
|
|
|
# update model info
|
|
if encrypted_credential.credential_info:
|
|
"""Update credential info"""
|
|
if "credential_info" not in merged_credential.credential_info:
|
|
merged_credential.credential_info = {}
|
|
merged_credential.credential_info.update(encrypted_credential.credential_info)
|
|
|
|
return merged_credential
|
|
|
|
|
|
@router.patch(
|
|
"/credentials/{credential_name}",
|
|
dependencies=[Depends(user_api_key_auth)],
|
|
tags=["credential management"],
|
|
)
|
|
async def update_credential(
|
|
request: Request,
|
|
fastapi_response: Response,
|
|
credential_name: str,
|
|
credential: CredentialItem,
|
|
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
|
|
):
|
|
"""
|
|
[BETA] endpoint. This might change unexpectedly.
|
|
"""
|
|
from litellm.proxy.proxy_server import prisma_client
|
|
|
|
try:
|
|
if prisma_client is None:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail={"error": CommonProxyErrors.db_not_connected_error.value},
|
|
)
|
|
db_credential = await prisma_client.db.litellm_credentialstable.find_unique(
|
|
where={"credential_name": credential_name},
|
|
)
|
|
if db_credential is None:
|
|
raise HTTPException(status_code=404, detail="Credential not found in DB.")
|
|
merged_credential = update_db_credential(db_credential, credential)
|
|
credential_object_jsonified = jsonify_object(merged_credential.model_dump())
|
|
await prisma_client.db.litellm_credentialstable.update(
|
|
where={"credential_name": credential_name},
|
|
data={
|
|
**credential_object_jsonified,
|
|
"updated_by": user_api_key_dict.user_id,
|
|
},
|
|
)
|
|
return {"success": True, "message": "Credential updated successfully"}
|
|
except Exception as e:
|
|
return handle_exception_on_proxy(e)
|