"""Simple API key and in-memory rate limiting.""" from __future__ import annotations import time from collections import defaultdict, deque from fastapi import Header, HTTPException from src.config import settings _request_windows: dict[str, deque[float]] = defaultdict(deque) def verify_api_key(x_api_key: str = Header(default="")): """Enforce API key when configured.""" if settings.api_key and x_api_key != settings.api_key: raise HTTPException(status_code=401, detail="Invalid API key") def check_rate_limit(client_id: str) -> None: """Allow at most N requests per minute for each client id.""" now = time.time() window_start = now - 60 q = _request_windows[client_id] while q and q[0] < window_start: q.popleft() if len(q) >= settings.rate_limit_per_minute: raise HTTPException(status_code=429, detail="Rate limit exceeded") q.append(now)