|
|
import time |
|
|
from collections import defaultdict |
|
|
|
|
|
class RateLimiter: |
|
|
def __init__(self): |
|
|
self.windows: dict[str, tuple[int, int]] = defaultdict(lambda: (0, 0)) |
|
|
|
|
|
def allow(self, ip: str, route: str, per_minute: int) -> bool: |
|
|
"""Checks if a request is allowed under a fixed-window rate limit.""" |
|
|
now = int(time.time()) |
|
|
current_window = now // 60 |
|
|
key = f"{ip}:{route}" |
|
|
|
|
|
window_start, count = self.windows.get(key, (0, 0)) |
|
|
|
|
|
if window_start != current_window: |
|
|
|
|
|
self.windows[key] = (current_window, 1) |
|
|
return True |
|
|
|
|
|
if count >= per_minute: |
|
|
|
|
|
return False |
|
|
|
|
|
|
|
|
self.windows[key] = (current_window, count + 1) |
|
|
return True |
|
|
|