spb/api-ka

1# ============================================2# Projet   : API-KA3# Fichier  : src/api/middleware/ratelimit.py4# Node     : m3u96b5# Author   : Simon-Pierre Boucher6# Contact  : contact@spboucher.ai7# Date     : 2026-08-168# ============================================9"""Rate limiting basique par IP (défaut : 120 req/min) — l'API est publique via ngrok."""1011from __future__ import annotations1213import time14from collections import defaultdict, deque1516from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint17from starlette.requests import Request18from starlette.responses import JSONResponse, Response1920from src.config import get_settings2122WINDOW_SECONDS = 60.023EXEMPT_PATHS = ("/health",)242526class RateLimitMiddleware(BaseHTTPMiddleware):27    """Fenêtre glissante de 60 s par IP ; au-delà de la limite → HTTP 429."""2829    def __init__(self, app, limit_per_minute: int | None = None) -> None:  # type: ignore[no-untyped-def]30        super().__init__(app)31        self.limit = limit_per_minute or get_settings().rate_limit_per_minute32        self._hits: dict[str, deque[float]] = defaultdict(deque)3334    async def dispatch(35        self, request: Request, call_next: RequestResponseEndpoint36    ) -> Response:37        if request.url.path in EXEMPT_PATHS:38            return await call_next(request)3940        client_ip = request.client.host if request.client else "unknown"41        now = time.monotonic()42        hits = self._hits[client_ip]43        while hits and now - hits[0] > WINDOW_SECONDS:44            hits.popleft()45        if len(hits) >= self.limit:46            return JSONResponse(47                status_code=429,48                content={49                    "success": False,50                    "error": f"Limite dépassée : {self.limit} requêtes/minute par IP",51                    "meta": {"node": get_settings().node_name},52                },53            )54        hits.append(now)55        return await call_next(request)56