mirror of
https://github.com/BerriAI/litellm.git
synced 2025-04-25 10:44:24 +00:00
45 lines
1.7 KiB
Python
45 lines
1.7 KiB
Python
from typing import Any, Dict
|
|
|
|
from fastapi import Request
|
|
|
|
from litellm._logging import verbose_proxy_logger
|
|
from litellm.proxy._types import UserAPIKeyAuth
|
|
|
|
|
|
async def handle_oauth2_proxy_request(request: Request) -> UserAPIKeyAuth:
|
|
"""
|
|
Handle request from oauth2 proxy.
|
|
"""
|
|
from litellm.proxy.proxy_server import general_settings
|
|
|
|
verbose_proxy_logger.debug("Handling oauth2 proxy request")
|
|
# Define the OAuth2 config mappings
|
|
oauth2_config_mappings: Dict[str, str] = general_settings.get(
|
|
"oauth2_config_mappings", None
|
|
)
|
|
verbose_proxy_logger.debug(f"Oauth2 config mappings: {oauth2_config_mappings}")
|
|
|
|
if not oauth2_config_mappings:
|
|
raise ValueError("Oauth2 config mappings not found in general_settings")
|
|
# Initialize a dictionary to store the mapped values
|
|
auth_data: Dict[str, Any] = {}
|
|
|
|
# Extract values from headers based on the mappings
|
|
for key, header in oauth2_config_mappings.items():
|
|
value = request.headers.get(header)
|
|
if value:
|
|
# Convert max_budget to float if present
|
|
if key == "max_budget":
|
|
auth_data[key] = float(value)
|
|
# Convert models to list if present
|
|
elif key == "models":
|
|
auth_data[key] = [model.strip() for model in value.split(",")]
|
|
else:
|
|
auth_data[key] = value
|
|
verbose_proxy_logger.debug(
|
|
f"Auth data before creating UserAPIKeyAuth object: {auth_data}"
|
|
)
|
|
user_api_key_auth = UserAPIKeyAuth(**auth_data)
|
|
verbose_proxy_logger.debug(f"UserAPIKeyAuth object created: {user_api_key_auth}")
|
|
# Create and return UserAPIKeyAuth object
|
|
return user_api_key_auth
|