Module genshin.client.ratelimit

Ratelimit handlers.

Functions

def handle_ratelimits(tries: int = 10,
exception: type[GenshinException] = genshin.errors.VisitsTooFrequently,
delay: float = 0.5) ‑> Callable[[~CallableT], ~CallableT]

Expand source code

def handle_ratelimits(
    tries: int = 10,
    exception: type[errors.GenshinException] = errors.VisitsTooFrequently,
    delay: float = 0.5,
) -> typing.Callable[[CallableT], CallableT]:
    """Handle ratelimits for requests."""
    return retry(
        stop=stop_after_attempt(tries),
        wait=wait_random_exponential(multiplier=delay, min=delay),
        retry=retry_if_exception_type(exception),
        reraise=True,
        before_sleep=before_sleep_log(LOGGER_, logging.DEBUG),
    )

Handle ratelimits for requests.

def handle_request_timeouts(tries: int = 10, delay: float = 0.5) ‑> Callable[[~CallableT], ~CallableT]

Expand source code

def handle_request_timeouts(
    tries: int = 10,
    delay: float = 0.5,
) -> typing.Callable[[CallableT], CallableT]:
    """Handle timeout errors for requests."""
    try:
        from aiohttp_socks import ProxyError
    except ImportError:
        return retry(
            stop=stop_after_attempt(tries),
            wait=wait_random_exponential(multiplier=delay, min=delay),
            retry=retry_if_exception_type((TimeoutError, aiohttp.ClientError)),
            reraise=True,
            before_sleep=before_sleep_log(LOGGER_, logging.DEBUG),
        )
    else:
        return retry(
            stop=stop_after_attempt(tries),
            wait=wait_random_exponential(multiplier=delay, min=delay),
            retry=retry_if_exception_type((TimeoutError, aiohttp.ClientError, ProxyError)),
            reraise=True,
            before_sleep=before_sleep_log(LOGGER_, logging.DEBUG),
        )

Handle timeout errors for requests.