22 lines
553 B
Python
22 lines
553 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:
|
|
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
|