Spaces:
Running
Running
File size: 2,113 Bytes
2c72e40 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 |
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()
|