API Reference

API Reference#

Main Interface#

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

Throttled class for synchronous rate limiting.

__call__(func)#

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[[ParamSpec(P, bound= None)], TypeVar(R)]

>>> 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, hooks=None)#

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 | str | None) – The quota for the rate limiter, default: 60 requests per minute. It accepts either: - throttled.rate_limiter.Quota - A quota DSL string, e.g. "100/s burst 200"
store (BaseStore | None) – 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.
hooks (Sequence[Hook] | None) – A sequence of hooks invoked by the middleware before and/or after each limit() operation, including any internal retries.

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.

Hooks#

class throttled.hooks.Hook#

Abstract base class for hooks using middleware pattern.

Custom hooks should inherit from this class and implement on_limit. The middleware pattern allows hooks to wrap the rate limit check, enabling timing measurement, exception handling, and more.

Example:

class MyHook(Hook):
    def on_limit(
        self,
        call_next: Callable[[], RateLimitResult],
        context: HookContext,
    ) -> RateLimitResult:
        start = time.time()
        result = call_next()
        elapsed = time.time() - start
        print(f"Key {context.key}: {elapsed:.3f}s, limited={result.limited}")
        return result

abstractmethod on_limit(call_next, context)#

Middleware that wraps a rate limit check.

Parameters:

call_next (Callable[[], RateLimitResult]) – Function to call the next hook or the actual rate limiter.
context (HookContext) – The rate-limiting context information.

Return type:

RateLimitResult

Returns:

The result from call_next() (RateLimitResult).

Observability#

FastAPI Integration#

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 the store and its backend.

Parameters:

server (str | None) – Store backend connection string.
options (dict[str, Any] | None) – Store backend options. For backend-specific configuration details, see Store Configuration.

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 the store and its backend.

Parameters:

server (str | None) – Store backend connection string.
options (dict[str, Any] | None) – Store backend options. For backend-specific configuration details, see Store Configuration.

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)#

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: RateLimitResult | None#: 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)#

Result produced by the rate limiter for a given key.

Exposes whether the request was limited, plus a lazily-materialized RateLimitState snapshot via the state property.

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

property state: RateLimitState#

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

Current state of the rate limiter for a 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#

Hooks#

Observability#

FastAPI Integration#

Store#

Rate Limiting#

Exceptions#

Lower-Level Classes#