test2 / app /services /rate_limiter.py
joedown11's picture
Upload 47 files
3910b9e verified
Raw
History Blame Contribute Delete
680 Bytes
import time
from collections import defaultdict, deque
from typing import Deque, Dict
class InMemoryRateLimiter:
def __init__(self, max_requests: int, window_seconds: int) -> None:
self.max_requests = max_requests
self.window_seconds = window_seconds
self._hits: Dict[str, Deque[float]] = defaultdict(deque)
def allow(self, key: str) -> bool:
now = time.time()
window_start = now - self.window_seconds
q = self._hits[key]
while q and q[0] < window_start:
q.popleft()
if len(q) >= self.max_requests:
return False
q.append(now)
return True