File size: 4,182 Bytes
54eb2ce | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | """
Utility functions for API key management and extraction.
"""
from typing import Optional, Dict
from fastapi import Request
def get_api_key_for_provider(
request: Request, provider: str, decrypted_keys: Optional[Dict[str, str]] = None
) -> Optional[str]:
"""
Get the API key for a specific provider from decrypted keys (user settings).
This function checks:
1. User-provided decrypted API keys from request (if passed as parameter)
2. Request state decrypted API keys (set by middleware)
User-provided API keys (gemini, groq, openrouter) should ONLY come from user settings,
not from environment variables.
Args:
request: FastAPI Request object
provider: The provider name (e.g., 'gemini', 'groq', 'openrouter')
decrypted_keys: Optional dictionary of decrypted keys (overrides request.state)
Returns:
The API key for the provider, or None if not found
"""
# Provider slug mapping to handle variations
provider_slug_map = {
"gemini": ["gemini", "google", "google-generativeai"],
"groq": ["groq"],
"openrouter": ["openrouter"],
}
provider = provider.lower()
possible_slugs = provider_slug_map.get(provider, [provider])
if decrypted_keys:
for slug in possible_slugs:
if slug in decrypted_keys:
return decrypted_keys[slug]
if hasattr(request.state, "decrypted_api_keys"):
for slug in possible_slugs:
if slug in request.state.decrypted_api_keys:
api_key = request.state.decrypted_api_keys[slug]
return api_key
else:
raise ValueError(
"request.state.decrypted_api_keys is not set. Ensure APIKeyDecryptionMiddleware is properly configured."
)
return None
def get_api_key_for_service(
request: Optional[Request], service_slug: str, env_var_name: str = None
) -> Optional[str]:
"""
Get API key for a user-provided service (embedding, web search, reranking, etc.)
User-provided services (cohere, tavily, firecrawl, langsearch) should ONLY come from user settings,
not from environment variables.
Args:
request: FastAPI Request object (can be None)
service_slug: The service slug (e.g., 'cohere', 'tavily', 'firecrawl', 'langsearch')
env_var_name: Deprecated - no longer used. Only user settings are checked.
Returns:
The API key if found, None otherwise
Raises:
ValueError: If the API key is not found in user settings
"""
# Check request state (set by middleware with user's decrypted keys)
if request and hasattr(request.state, "decrypted_api_keys"):
api_key = request.state.decrypted_api_keys.get(service_slug)
if api_key:
return api_key
# No fallback to environment variables - user must provide the key in Settings
service_names = {
"cohere": "Cohere",
"tavily": "Tavily",
"firecrawl": "Firecrawl",
"langsearch": "LangSearch",
}
service_display = service_names.get(service_slug, service_slug)
raise ValueError(
f"No {service_display} API key found. Please add your {service_display} API key in Settings."
)
def extract_decrypted_keys_dict(api_keys_encrypted: Optional[list]) -> Dict[str, str]:
"""
Extract decrypted API keys from a list of encrypted API key items.
Note: This function expects the keys to already be decrypted by the middleware.
It simply converts the list format to a dictionary format.
Args:
api_keys_encrypted: List of API key items (already decrypted by middleware)
Returns:
Dictionary mapping service slug to API key
"""
if not api_keys_encrypted:
return {}
decrypted_keys = {}
for item in api_keys_encrypted:
if isinstance(item, dict):
slug = item.get("slug")
api_key = item.get("api_key")
if slug and api_key:
decrypted_keys[slug] = api_key
return decrypted_keys
|