API Reference

API Reference#

Main Interface#

class throttled.Throttled(key=None, timeout=None, using=None, quota=None, store=None, cost=1)#

Throttled class for synchronous rate limiting.

__call__(func=None)#

Decorator to apply rate limiting to a function.

The cost value is taken from the Throttled instance’s initialization.

Usage:

>>> from throttled import Throttled
>>>
>>> @Throttled(key="key")
>>> def demo(): pass

or with cost:

Return type:: Callable | Callable[[Callable], Callable]

>>> from throttled import Throttled
>>>
>>> @Throttled(key="key", cost=2)
>>> def demo(): pass

__enter__()#

Context manager to apply rate limiting to a block of code.

Return type:: RateLimitResult
Returns:: RateLimitResult - The result of the rate limiting check.
Raise:: throttled.exceptions.LimitedError if the rate limit is exceeded.

__init__(key=None, timeout=None, using=None, quota=None, store=None, cost=1)#

Initializes the Throttled class.

Parameters:

key (str | None) – The unique identifier for the rate limit subject, e.g. user ID or IP address.
timeout (float | None) – Maximum wait time in seconds when rate limit is exceeded. (Default) If set to -1, it will return immediately. Otherwise, it will block until the request can be processed or the timeout is reached.
using (str | None) – The type of rate limiter to use, you can choose from RateLimiterType, default: token_bucket.
quota (Quota | None) – The quota for the rate limiter, default: 60 requests per minute.
store (throttled.store.BaseStore) – The store to use for the rate limiter. By default, it uses the global shared throttled.store.MemoryStore instance with maximum capacity of 1024, so you don’t usually need to create it manually.
cost (int) – The cost of each request in terms of how much of the rate limit quota it consumes, default: 1.

limit(key=None, cost=1, timeout=None)#

Apply rate limiting logic to a given key with a specified cost.

Parameters:

key (str | None) – The unique identifier for the rate limit subject, e.g. user ID or IP address, it will override the instance key if provided.
cost (int) – The cost of the current request in terms of how much of the rate limit quota it consumes.
timeout (float | None) –
Maximum wait time in seconds when rate limit is exceeded, overrides the instance timeout if provided. When invoked with the timeout argument set to a positive float (defaults to -1, which means return immediately):
- If timeout < RateLimitState.retry_after, it will return immediately.
- If timeout >= RateLimitState.retry_after, it will block until the request can be processed or the timeout is reached.

Return type:

RateLimitResult

Returns:

The result of the rate limiting check.

Raise:

throttled.exceptions.DataError if invalid parameters are provided.

peek(key)#

Retrieve the current state of rate limiter for the given key.

Parameters:: key (str) – The unique identifier for the rate limit subject, e.g. user ID or IP address.
Return type:: RateLimitState
Returns:: throttled.RateLimitState - The current state of the rate limiter for the given key.

Store#

class throttled.store.BaseStore(server=None, options=None)#: Abstract class for all stores.

class throttled.store.MemoryStore(server=None, options=None)#

Bases: BaseStore

Concrete implementation of BaseStore using Memory as backend.

throttled.store.MemoryStore is essentially a memory-based LRU Cache with expiration time, it is thread-safe and can be used for rate limiting in a single process.

__init__(server=None, options=None)#: Initialize MemoryStore, see MemoryStore Arguments.

class throttled.store.RedisStore(server=None, options=None)#

Bases: BaseStore

Concrete implementation of BaseStore using Redis as backend.

throttled.store.RedisStore is implemented based on redis-py, you can use it for rate limiting in a distributed environment.

__init__(server=None, options=None)#

Initialize RedisStore.

Parameters:

server (str | None) – Redis Standard Redis URL, you can use it to connect to Redis in any deployment mode, see Store Backends.
options (dict[str, Any] | None) –
Redis connection configuration, supports all configuration item of redis-py, see RedisStore Options.

Rate Limiting#

class throttled.rate_limiter.Rate(period, limit)#

Rate represents the rate limit configuration.

limit: int#: The maximum number of requests allowed within the specified period.

period: timedelta#: The time period for which the rate limit applies.

class throttled.rate_limiter.Quota(rate, burst=0, period_sec=None, emission_interval=None, fill_rate=None)#

Quota represents the quota limit configuration.

burst: int = 0#: Optional burst capacity that allows exceeding the rate limit momentarily. Default is 0, which means no burst capacity.

rate: Rate#: The base rate limit configuration.

throttled.rate_limiter.per_sec(limit, burst=None)#

Create a quota representing the maximum requests and burst per second.

Return type:: Quota

throttled.rate_limiter.per_min(limit, burst=None)#

Create a quota representing the maximum requests and burst per minute.

Return type:: Quota

throttled.rate_limiter.per_hour(limit, burst=None)#

Create a quota representing the maximum requests and burst per hour.

Return type:: Quota

throttled.rate_limiter.per_day(limit, burst=None)#

Create a quota representing the maximum requests and burst per day.

Return type:: Quota

throttled.rate_limiter.per_week(limit, burst=None)#

Create a quota representing the maximum requests and burst per week.

Return type:: Quota

throttled.rate_limiter.per_duration(duration, limit, burst=None)#

Create a quota representing the maximum requests and burst per duration.

Return type:: Quota

Exceptions#

All exceptions inherit from throttled.exceptions.BaseThrottledError.

exception throttled.exceptions.BaseThrottledError#: Base class for all throttled-related exceptions.

exception throttled.exceptions.SetUpError#: Exception raised when there is an error during setup.

exception throttled.exceptions.DataError#

Exception raised for errors related to data integrity or format.

Thrown when the parameter is invalid, such as: Invalid key: None, must be a non-empty key.

exception throttled.exceptions.StoreUnavailableError#: Exception raised when the store (e.g., Redis) is unavailable.

exception throttled.exceptions.LimitedError(rate_limit_result=None)#

Exception raised when a rate limit is exceeded.

When a request is throttled, an exception is thrown, such as: Rate limit exceeded: remaining=0, reset_after=60, retry_after=60.

rate_limit_result: Optional[RateLimitResult]#: The result after executing the RateLimiter for the given key.

Lower-Level Classes#

class throttled.RateLimiterType(*values)#

Enumeration for types of RateLimiter.

FIXED_WINDOW = 'fixed_window'#

GCRA = 'gcra'#

LEAKING_BUCKET = 'leaking_bucket'#

SLIDING_WINDOW = 'sliding_window'#

TOKEN_BUCKET = 'token_bucket'#

class throttled.RateLimitResult(limited, state_values)#

RateLimitResult represents the result after executing the RateLimiter for the given key.

limited: bool#: Represents whether this request is allowed to pass.

property state: RateLimitState#

class throttled.RateLimitState(limit, remaining, reset_after, retry_after=0)#

RateLimitState represents the current state of the rate limiter for the given key.

limit: int#: Represents the maximum number of requests allowed to pass in the initial state.

remaining: int#: Represents the maximum number of requests allowed to pass for the given key in the current state.

reset_after: float#: Represents the time in seconds for the RateLimiter to return to its initial state. In the initial state, limit = remaining.

retry_after: float = 0#: Represents the time in seconds for the request to be retried, 0 if the request is allowed.

API Reference

Contents

API Reference#

Main Interface#

Store#

Rate Limiting#

Exceptions#

Lower-Level Classes#