42 lines
1.2 KiB
Python
42 lines
1.2 KiB
Python
"""Простой in-memory rate limit (на процесс)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from collections import defaultdict
|
|
from threading import Lock
|
|
|
|
|
|
class RateLimiter:
|
|
def __init__(self) -> None:
|
|
self._hits: dict[str, list[float]] = defaultdict(list)
|
|
self._lock = Lock()
|
|
|
|
def hit(self, key: str, *, limit: int, window_sec: float) -> bool:
|
|
"""True если запрос разрешён, False если превышен лимит."""
|
|
now = time.monotonic()
|
|
with self._lock:
|
|
bucket = self._hits[key]
|
|
cutoff = now - window_sec
|
|
self._hits[key] = [t for t in bucket if t >= cutoff]
|
|
if len(self._hits[key]) >= limit:
|
|
return False
|
|
self._hits[key].append(now)
|
|
return True
|
|
|
|
|
|
limiter = RateLimiter()
|
|
|
|
|
|
def client_ip(request) -> str:
|
|
forwarded = (request.headers.get("X-Forwarded-For") or "").split(",")[0].strip()
|
|
if forwarded:
|
|
return forwarded
|
|
peer = request.remote
|
|
return str(peer or "unknown")
|
|
|
|
|
|
def allow(request, bucket: str, *, limit: int = 20, window_sec: float = 60.0) -> bool:
|
|
ip = client_ip(request)
|
|
return limiter.hit(f"{bucket}:{ip}", limit=limit, window_sec=window_sec)
|