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