from fastapi import Security, HTTPException, status, Depends from fastapi.security.api_key import APIKeyHeader from sqlalchemy.orm import Session from app.core.config import settings from app.core.database import get_db from app.models.module import Module from cachetools import TTLCache api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False) # Cache keys for 5 minutes, store up to 1000 keys # This prevents a Supabase round-trip on every message auth_cache = TTLCache(maxsize=1000, ttl=300) async def get_api_key( api_key_header: str = Security(api_key_header), db: Session = Depends(get_db) ): if not api_key_header: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="API Key missing" ) # 1. Fallback to global static key (Admin) if api_key_header == settings.API_KEY: return api_key_header # 2. Check Cache first (VERY FAST) if api_key_header in auth_cache: return api_key_header # 3. Check Database for Module key (Database round-trip) module = db.query(Module).filter(Module.secret_key == api_key_header, Module.is_active == True).first() if module: # Save module ID to cache for next time auth_cache[api_key_header] = module.id return module raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Could not validate credentials or API Key is inactive" ) async def get_current_module( api_key_header: str = Security(api_key_header), db: Session = Depends(get_db) ): # 1. Fallback to global static key (Admin) - No module tracking if api_key_header == settings.API_KEY: return None # 2. Check Cache if api_key_header in auth_cache: module_id = auth_cache[api_key_header] return db.query(Module).filter(Module.id == module_id).first() # 3. DB Lookup module = db.query(Module).filter(Module.secret_key == api_key_header, Module.is_active == True).first() if module: auth_cache[api_key_header] = module.id return module raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Could not validate credentials" )