"""Rate limit base throttling.""" from typing import TYPE_CHECKING from .base import BaseThrottler if TYPE_CHECKING: from ..client.base import BaseExtendedClient __all__ = ["RateLimitThrottler"] class RateLimitThrottler(BaseThrottler): """Throttle based on rate limit information.""" def __init__(self, threshold_factor: float = 0.5, scale_factor: float = 1, *args, **kwargs): """Initialize RateLimitThrottler. :param threshold_factor: Factor representing the portion of the limit considered as the safe zone. When remaining requests are within this portion, no sleep is needed. Must be within [0, 1]. i.e. 0.75 means 75% of the limit. :param scale_factor: Factor to scale the calculated sleep time, making the sleep time function. Must be greater than 0 """ super().__init__(*args, **kwargs) if not (0 <= threshold_factor <= 1): raise ValueError("threshold_factor must be within the range [0, 1]") if scale_factor <= 0: raise ValueError("scale_factor must be greater than 0") self._threshold_factor = threshold_factor self._scale_factor = scale_factor def _get_sleep_time(self, client: "BaseExtendedClient") -> float: return self._get_sleep_time_on_rate_limit( client.ratelimit_info.remaining, client.ratelimit_info.limit, client.ratelimit_info.reset_in_seconds ) def _get_sleep_time_on_rate_limit(self, remaining: int, limit: int, reset_in_seconds: int) -> float: if remaining >= limit * (1 - self._threshold_factor): sleep_time = 0.0 elif remaining > 0: # Calculate allowed requests per second allowed_rps = remaining / reset_in_seconds if reset_in_seconds > 0 else 0 # Calculate sleep time needed to stay within the rate limit sleep_time = 1 / allowed_rps if allowed_rps > 0 else reset_in_seconds # Multiply sleep time by scale factor to make sleep time function more responsive sleep_time = sleep_time * self._scale_factor # Ensure sleep time is within reasonable bounds sleep_time = max(0.0, min(sleep_time, reset_in_seconds)) else: # If no requests are remaining, sleep until the rate limit resets sleep_time = reset_in_seconds return round(sleep_time, 2)