api/security.py

"""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)