28 lines
776 B
Python
28 lines
776 B
Python
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
from . import settings
|
|
|
|
_RATE_BUCKETS: dict[str, dict[str, list[float]]] = {}
|
|
|
|
|
|
def rate_limit_allow(ip: str, *, key: str, limit: int, window_sec: int) -> bool:
|
|
"""Return whether a request bucket still has capacity.
|
|
|
|
WHY: access-request endpoints need a simple in-process guard that is easy to
|
|
exercise in tests and cheap to apply before any heavier work starts.
|
|
"""
|
|
|
|
if limit <= 0:
|
|
return True
|
|
now = time.time()
|
|
window_start = now - window_sec
|
|
buckets_by_ip = _RATE_BUCKETS.setdefault(key, {})
|
|
bucket = buckets_by_ip.setdefault(ip, [])
|
|
bucket[:] = [t for t in bucket if t >= window_start]
|
|
if len(bucket) >= limit:
|
|
return False
|
|
bucket.append(now)
|
|
return True
|