| """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() |