Decorators

High level API for limiting callbacks with static rates.

@uprate.decorators.ratelimit(rate: Rate | RateGroup, *, key: Callable[..., Key] | Callable[..., Coroutine[Any, Any, Key]] | None = None, on_retry: Callable[[RateLimitError], Any] | Callable[[RateLimitError], Coroutine] | None = None, store: BaseStore | SyncStore | None = None)[source]

Limit a coroutine function or a callable to be called within provided rate. Example here

Parameters

  • rate (Rate, RateGroup, (Rate | RateGroup)) – The rate that the decorated function must follow.

  • key (Callable[…, Key] | Callable[…, Coroutine[Any, Any, Key]] | None) – The callback for generating a bucket for ratelimit from the arguments provided to the decorated function, this can be a coroutine function only when the decorated is a coroutine function as well. If None, a default callback returning a string based on the decorated function’s name is used, by default None.

  • on_retry (Callable[[RateLimitError], Any] | Callable[[RateLimitError], Coroutine] | None) – If provided then this function will be called when the function gets ratelimited and then the decorated function will be called again, if None then function call isn’t retried and RateLimitError is raised, by default None.

  • store (BaseStore | SyncStore | None) – The store to use for the rate limit, must be of type BaseStore if decorated function is a coroutine function else SyncStore. If None then a suitable derived memory store is used, by default None.

Raises

.RateLimitError – on_retry parameter is not provided and the decorated function got ratelimited.

async uprate.decorators.on_retry_sleep(error: uprate.errors.RateLimitError) → None[source]

Make the current task yield to the event_loop till a usage token is available for the rate limit which raised uprate.errors.RateLimitError

Parameters

error (RateLimitError) – The rate limit error to sleep for.

Note

uprate.decorators.on_retry_sleep() can be used with uprate.decorators.ratelimit() only when the decorated function is async.

uprate.decorators.on_retry_block(error: uprate.errors.RateLimitError) → None[source]

Block the current thread till a usage token is available for the rate limit which raised uprate.errors.RateLimitError

Parameters

error (RateLimitError) – The rate limit error to block for.