"""Helper functions to handle rate limit headers.""" import time from typing import Any, Mapping from .. import config __all__ = ["RatelimitInfo"] class RatelimitInfo: """Container for ratelimit data.""" _default_limit = 1000 _default_remaining = 1000 _default_reset_ts = 0 def __init__(self: "RatelimitInfo"): """Initialize RatelimitInfo.""" self._limit: int = self._default_limit self._remaining: int = self._default_remaining self._reset_ts: int = self._default_reset_ts @staticmethod def _parse_int(value: str, default: int = 0) -> int: try: result = int(value) except (ValueError, TypeError): result = default return result def set_from_headers(self, headers: Mapping[str, str]): """Set ratelimit info from headers. :param headers: Mapping of headers :return: """ self._limit = self._parse_int(headers.get(config.RATELIMIT_LIMIT_HEADER, ""), self._default_limit) self._remaining = self._parse_int(headers.get(config.RATELIMIT_REMAINING_HEADER, ""), self._default_remaining) self._reset_ts = self._parse_int(headers.get(config.RATELIMIT_RESET_HEADER, ""), self._default_reset_ts) @property def limit(self) -> int: """Total number of requests per minute.""" return self._limit @property def remaining(self) -> int: """Number of remaining requests per minute.""" return self._remaining @property def reset_ts(self): """Timestamp when the request limit will reset.""" return self._reset_ts @property def reset_in_seconds(self) -> int: """Seconds until next request limit reset.""" return max(0, self._reset_ts - int(time.time())) def to_dict(self) -> dict[str, Any]: """Convert self to a dictionary. :return: dictionary representation of the rate limit info. """ return { "ratelimit_limit": self.limit, "ratelimit_remaining": self.remaining, "ratelimit_reset": self.reset_ts, "ratelimit_reset_in_seconds": self.reset_in_seconds, }