feat: add rate limiter utility
This commit is contained in:
22
backend/app/rate_limit.py
Normal file
22
backend/app/rate_limit.py
Normal file
@@ -0,0 +1,22 @@
|
||||
import time
|
||||
from collections import defaultdict
|
||||
|
||||
|
||||
class RateLimiter:
|
||||
"""Fixed-window limiter keyed by an arbitrary string (user id or client IP)."""
|
||||
|
||||
def __init__(self, max_requests: int, window_seconds: float):
|
||||
self.max_requests = max_requests
|
||||
self.window_seconds = window_seconds
|
||||
self._hits: dict[str, list[float]] = defaultdict(list)
|
||||
|
||||
def allow(self, key: str) -> bool:
|
||||
now = time.monotonic()
|
||||
window_start = now - self.window_seconds
|
||||
hits = self._hits[key]
|
||||
while hits and hits[0] < window_start:
|
||||
hits.pop(0)
|
||||
if len(hits) >= self.max_requests:
|
||||
return False
|
||||
hits.append(now)
|
||||
return True
|
||||
16
backend/tests/test_rate_limit.py
Normal file
16
backend/tests/test_rate_limit.py
Normal file
@@ -0,0 +1,16 @@
|
||||
from app.rate_limit import RateLimiter
|
||||
|
||||
|
||||
def test_allows_up_to_limit_then_blocks():
|
||||
limiter = RateLimiter(max_requests=3, window_seconds=60)
|
||||
assert limiter.allow("user-1") is True
|
||||
assert limiter.allow("user-1") is True
|
||||
assert limiter.allow("user-1") is True
|
||||
assert limiter.allow("user-1") is False
|
||||
|
||||
|
||||
def test_different_keys_tracked_independently():
|
||||
limiter = RateLimiter(max_requests=1, window_seconds=60)
|
||||
assert limiter.allow("user-1") is True
|
||||
assert limiter.allow("user-2") is True
|
||||
assert limiter.allow("user-1") is False
|
||||
Reference in New Issue
Block a user