File size: 604 Bytes
23d8f88 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 | """Rate limiting for requests."""
import time
from typing import Dict
from threading import Lock
_rate_limits: Dict[str, float] = {}
_lock = Lock()
def check_rate_limit(key: str, min_interval: float = 1.0):
"""Check and enforce rate limit."""
with _lock:
now = time.time()
if key in _rate_limits:
elapsed = now - _rate_limits[key]
if elapsed < min_interval:
time.sleep(min_interval - elapsed)
_rate_limits[key] = time.time()
def reset_rate_limits():
"""Reset all rate limits."""
with _lock:
_rate_limits.clear() |