Spaces:
Runtime error
Runtime error
File size: 3,666 Bytes
d87d194 |
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 |
import os
import time
from dotenv import load_dotenv
from .supabase_cli import get_supabase_client # Import from the existing supabase_cli
load_dotenv()
# Simple in-memory cache
api_key_cache = {}
last_loaded_time = 0
CACHE_TTL = 300 # Cache time-to-live in seconds (5 minutes)
def load_api_keys_from_db():
"""Loads API keys from the database into the cache."""
global api_key_cache
print("Loading API keys from database...")
supabase = get_supabase_client() # Use the imported function
if not supabase:
return {}
try:
response = supabase.from_("app_config").select("config").eq("id", 1).execute()
#[
# {
# 'config': {
# 'api_keys': [
# {'api_key': 'xxxxx', 'can_use_at': 0},
# {'api_key': 'yyyyy', 'can_use_at': 0}
# ]
# }
# }
#]
# print(f"Supabase response data: {response.data}")
if response.data and response.data[0] and 'config' in response.data[0] and 'api_keys' in response.data[0]['config']:
api_keys_list = response.data[0]['config']['api_keys']
if api_keys_list:
api_key_cache = {item['api_key']: item['can_use_at'] for item in api_keys_list}
# {'xxxxx': 0, 'yyyyy': 0}
# print(f"Successfully loaded {len(api_key_cache)} API keys with can_use_at timestamps.")
# print(f"API keys cache: {api_key_cache}")
else:
print("API keys list is empty in the database response.")
api_key_cache = {}
else:
print("Unexpected database response structure or no data found.")
api_key_cache = {}
except Exception as e:
print(f"Error loading API keys from database: {e}")
api_key_cache = {} # Clear cache on error
return api_key_cache
def get_api_key():
"""Gets an API key from the cache, loading from DB if necessary."""
global api_key_cache
# print("\n\n\n\n\nGetting API key...")
# print(api_key_cache)
# Ensure cache is loaded
if not api_key_cache :
load_api_keys_from_db()
# api_key_cache={'xxxxx': 1, 'yyyyy': 0}
# print('api_key_cache',api_key_cache)
# Sort available keys by their 'can_use_at' timestamp in ascending order
sorted_key_cache = sorted(api_key_cache, key=lambda key: api_key_cache.get(key, float('inf')))
# print('\n\n\n\n\n\n\n\n\nsorted_key_cache',sorted_key_cache)
# Select the key with the smallest 'can_use_at' value
selected_key = sorted_key_cache[0]
# TODO: Update the 'can_use_at' timestamp for the selected key in the cache
# This requires knowing the rate limit or delay period for the API.
# For now, we'll just return the key. In a real system, you'd update the cache
# with a future timestamp after using the key successfully.
# Example: api_key_cache[selected_key] = time.time() + API_RATE_LIMIT_DELAY
# print(f"Using API key: {selected_key[:5]}...") # Print partial key for security
api_key_cache[selected_key]=time.time()
return selected_key
# Example usage (for testing)
if __name__ == "__main__":
# In a real application, environment variables would be set
# For testing, you might set them here or use a .env file loader
# os.environ['SUPABASE_URL'] = 'your_supabase_url'
# os.environ['SUPABASE_KEY'] = 'your_supabase_key'
# keys = load_api_keys_from_db()
# print("Loaded keys:", keys)
# # Simulate using cached keys
# keys = load_api_keys_from_db()
# print("Loaded keys (from cache):", keys)
pass
|