Spaces:
Running
Running
| 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 | |