app/rate_limit.py

from collections import deque
from threading import Lock
from time import time

from fastapi import HTTPException, Request, status

from app.config import settings

_hits: dict[str, deque[float]] = {}
_lock = Lock()


def _client_ip(request: Request) -> str:
    forwarded = request.headers.get("x-forwarded-for")
    if forwarded:
        return forwarded.split(",")[0].strip()
    if request.client:
        return request.client.host
    return "unknown"


def enforce_rate_limit(request: Request, bucket: str) -> None:
    now = time()
    window = settings.AUTH_RATE_LIMIT_WINDOW_SECONDS
    limit = settings.AUTH_RATE_LIMIT_MAX_REQUESTS
    key = f"{bucket}:{_client_ip(request)}"

    with _lock:
        q = _hits.setdefault(key, deque())
        cutoff = now - window
        while q and q[0] < cutoff:
            q.popleft()
        if len(q) >= limit:
            raise HTTPException(
                status_code=status.HTTP_429_TOO_MANY_REQUESTS,
                detail="Too many requests. Try again later.",
            )
        q.append(now)