37 lines
1.1 KiB
Python
37 lines
1.1 KiB
Python
"""
|
|
BAN YARO — Rate Limiter
|
|
Sliding-Window-Limiter, in-memory (kein Redis nötig für Single-Container).
|
|
"""
|
|
|
|
import threading
|
|
from collections import defaultdict, deque
|
|
from datetime import datetime, timedelta
|
|
|
|
from fastapi import HTTPException, Request
|
|
|
|
_buckets: dict[str, deque] = defaultdict(deque)
|
|
_lock = threading.Lock()
|
|
|
|
|
|
def check(request: Request, *, max_requests: int, window_seconds: int, key: str = ""):
|
|
"""
|
|
Wirft HTTP 429 wenn max_requests im Zeitfenster überschritten.
|
|
key: optionaler Präfix um verschiedene Limits zu trennen (z.B. 'register', 'login').
|
|
"""
|
|
ip = (request.client.host if request.client else "unknown")
|
|
bucket_key = f"{key}:{ip}"
|
|
now = datetime.utcnow()
|
|
cutoff = now - timedelta(seconds=window_seconds)
|
|
|
|
with _lock:
|
|
dq = _buckets[bucket_key]
|
|
# Alte Einträge raus
|
|
while dq and dq[0] < cutoff:
|
|
dq.popleft()
|
|
if len(dq) >= max_requests:
|
|
minutes = window_seconds // 60
|
|
raise HTTPException(
|
|
429,
|
|
f"Zu viele Versuche. Bitte warte {minutes} Minute(n) und versuche es erneut."
|
|
)
|
|
dq.append(now)
|