Spaces:
Sleeping
Sleeping
| """🏆 Rate Limiter for API Protection""" | |
| import time | |
| from collections import defaultdict | |
| from typing import Dict, Tuple | |
| import logging | |
| logger = logging.getLogger(__name__) | |
| class RateLimiter: | |
| """Token bucket rate limiter""" | |
| def __init__(self, requests_per_minute: int = 60): | |
| self.capacity = requests_per_minute | |
| self.refill_rate = requests_per_minute / 60.0 # Tokens per second | |
| self.buckets: Dict[str, Tuple[float, float]] = defaultdict(lambda: (self.capacity, time.time())) | |
| def check_rate_limit(self, identifier: str = "default") -> bool: | |
| """ | |
| Check if request is allowed under rate limit. | |
| Args: | |
| identifier: User/IP identifier for rate limiting | |
| Returns: | |
| bool: True if request allowed, False if rate limited | |
| """ | |
| tokens, last_update = self.buckets[identifier] | |
| now = time.time() | |
| # Refill tokens based on time elapsed | |
| time_passed = now - last_update | |
| tokens = min(self.capacity, tokens + time_passed * self.refill_rate) | |
| if tokens >= 1: | |
| # Consume 1 token | |
| self.buckets[identifier] = (tokens - 1, now) | |
| return True | |
| else: | |
| # Rate limited | |
| self.buckets[identifier] = (tokens, now) | |
| logger.warning(f"Rate limit exceeded for {identifier}") | |
| return False | |
| def get_wait_time(self, identifier: str = "default") -> float: | |
| """Get time to wait before next request allowed""" | |
| tokens, _ = self.buckets[identifier] | |
| if tokens >= 1: | |
| return 0.0 | |
| return (1 - tokens) / self.refill_rate | |
| # Global rate limiter instance | |
| rate_limiter = RateLimiter(requests_per_minute=60) | |