Sergey Antropoff
30 lines
797 B
Python
30 lines
797 B
Python
from __future__ import annotations
|
|
|
|
import time
|
|
from collections import defaultdict, deque
|
|
from threading import Lock
|
|
|
|
|
|
class RateLimiter:
|
|
"""In-memory sliding window limiter (per process). Good enough for single-replica / dev."""
|
|
|
|
def __init__(self) -> None:
|
|
self._hits: dict[str, deque[float]] = defaultdict(deque)
|
|
self._lock = Lock()
|
|
|
|
def allow(self, key: str, limit: int, window_seconds: int = 60) -> bool:
|
|
if limit <= 0:
|
|
return True
|
|
now = time.monotonic()
|
|
with self._lock:
|
|
q = self._hits[key]
|
|
while q and now - q[0] > window_seconds:
|
|
q.popleft()
|
|
if len(q) >= limit:
|
|
return False
|
|
q.append(now)
|
|
return True
|
|
|
|
|
|
rate_limiter = RateLimiter()
|