spb/uqo-chat

1"""Sliding-window rate limiter. Redis if configured, otherwise in-process memory."""23from __future__ import annotations45import time6from collections import defaultdict, deque78from fastapi import HTTPException, status91011class MemoryRateLimiter:12    def __init__(self) -> None:13        self._hits: dict[str, deque[float]] = defaultdict(deque)1415    def check(self, key: str, limit: int, window_s: int, message_fr: str) -> None:16        now = time.time()17        q = self._hits[key]18        while q and q[0] < now - window_s:19            q.popleft()20        if len(q) >= limit:21            raise HTTPException(status.HTTP_429_TOO_MANY_REQUESTS, detail=message_fr)22        q.append(now)2324    def count(self, key: str, window_s: int) -> int:25        now = time.time()26        q = self._hits[key]27        while q and q[0] < now - window_s:28            q.popleft()29        return len(q)303132limiter = MemoryRateLimiter()3334MSG_MESSAGES = "Tu as atteint la limite de messages pour l'instant. Réessaie un peu plus tard."35MSG_SANDBOX = "Limite d'exécutions Python atteinte pour cette heure."36MSG_WEB = "Limite de recherches web atteinte pour aujourd'hui."37MSG_UPLOAD = "Limite de téléversements atteinte pour aujourd'hui."38