Spaces:
Running
Running
| import time | |
| from collections import deque | |
| from typing import Deque, Tuple | |
| class RateLimiter: | |
| """ | |
| Simple rate limiter to prevent overwhelming target websites or APIs | |
| Implements a sliding window algorithm | |
| """ | |
| def __init__(self, window_size: int = 60, max_requests: int = 10): | |
| """ | |
| Initialize the rate limiter | |
| Args: | |
| window_size: Time window in seconds | |
| max_requests: Maximum allowed requests in the window | |
| """ | |
| self.window_size = window_size | |
| self.max_requests = max_requests | |
| self.request_timestamps: Deque[float] = deque() | |
| def record_request(self) -> None: | |
| """Record a request with the current timestamp""" | |
| current_time = time.time() | |
| self.request_timestamps.append(current_time) | |
| self._clean_old_timestamps(current_time) | |
| def can_proceed(self) -> bool: | |
| """Check if a new request can proceed within rate limits""" | |
| current_time = time.time() | |
| self._clean_old_timestamps(current_time) | |
| return len(self.request_timestamps) < self.max_requests | |
| def get_wait_time(self) -> float: | |
| """Calculate time to wait (in seconds) before next request is allowed""" | |
| if self.can_proceed(): | |
| return 0.0 | |
| current_time = time.time() | |
| oldest_allowed_time = current_time - self.window_size | |
| if not self.request_timestamps: | |
| return 0.0 | |
| # Find the oldest timestamp in the window and calculate when it will expire | |
| oldest_in_window = self.request_timestamps[0] | |
| time_until_oldest_expires = oldest_in_window - oldest_allowed_time | |
| return max(0.0, time_until_oldest_expires) | |
| def _clean_old_timestamps(self, current_time: float) -> None: | |
| """Remove timestamps that are outside the current window""" | |
| oldest_allowed_time = current_time - self.window_size | |
| while self.request_timestamps and self.request_timestamps[0] < oldest_allowed_time: | |
| self.request_timestamps.popleft() | |