rate_limit

class litestar.middleware.rate_limit.CacheObject

Bases: object

Representation of a cached object’s metadata.

__init__(history: list[int], reset: int) → None

class litestar.middleware.rate_limit.RateLimitConfig

Bases: object

Configuration for RateLimitMiddleware

rate_limit: tuple[DurationUnit, int]

A tuple containing a time unit (second, minute, hour, day) and quantity, e.g. (“day”, 1) or (“minute”, 5).

exclude: str | list[str] | None = None

A pattern or list of patterns to skip in the rate limiting middleware.

exclude_opt_key: str | None = None

An identifier to use on routes to disable rate limiting for a particular route.

identifier_for_request() → str

A callable that receives the request and returns an identifier for which the limit should be applied. Defaults to get_remote_address(), which returns the client’s address.

Note that get_remote_address() does NOT honour X-FORWARDED-FOR headers, as these cannot be trusted implicitly. If running behind a proxy, a secure way of updating the client’s address should be implemented, such as uvicorn’s ProxyHeaderMiddleware or hypercon’s ProxyFixMiddleware .

check_throttle_handler: Callable[[Request[Any, Any, Any]], SyncOrAsyncUnion[bool]] | None = None

Handler callable that receives the request instance, returning a boolean dictating whether or not the request should be checked for rate limiting.

middleware_class

The middleware class to use.

alias of RateLimitMiddleware

set_rate_limit_headers: bool = True

Boolean dictating whether to set the rate limit headers on the response.

__init__(rate_limit: tuple[DurationUnit, int], exclude: str | list[str] | None = None, exclude_opt_key: str | None = None, identifier_for_request: Callable[[Request], str] = <function get_remote_address>, check_throttle_handler: Callable[[Request[Any, Any, Any]], SyncOrAsyncUnion[bool]] | None = None, middleware_class: type[RateLimitMiddleware] = <class 'litestar.middleware.rate_limit.RateLimitMiddleware'>, set_rate_limit_headers: bool = True, rate_limit_policy_header_key: str = 'RateLimit-Policy', rate_limit_remaining_header_key: str = 'RateLimit-Remaining', rate_limit_reset_header_key: str = 'RateLimit-Reset', rate_limit_limit_header_key: str = 'RateLimit-Limit', store: str = 'rate_limit') → None

rate_limit_policy_header_key: str = 'RateLimit-Policy'

Key to use for the rate limit policy header.

rate_limit_remaining_header_key: str = 'RateLimit-Remaining'

Key to use for the rate limit remaining header.

rate_limit_reset_header_key: str = 'RateLimit-Reset'

Key to use for the rate limit reset header.

rate_limit_limit_header_key: str = 'RateLimit-Limit'

Key to use for the rate limit limit header.

store: str = 'rate_limit'

Name of the Store to use

property middleware: RateLimitMiddleware

Create an instance of middleware_class, configured from this config instance.

Deprecated since version 3.0: Construct a RateLimitMiddleware instance directly and pass it to the middleware list instead, e.g. middleware=[RateLimitMiddleware(rate_limit=("minute", 10), exclude=["/schema"])].

Returns:

An instance of middleware_class, configured from this config instance.

get_store_from_app(app: Litestar) → Store

Get the store defined in store from an Litestar instance.

class litestar.middleware.rate_limit.RateLimitMiddleware

Bases: ASGIMiddleware

Rate-limiting middleware.

scopes: tuple[ScopeType, ...] = (ScopeType.HTTP, ScopeType.ASGI)

Scope types this middleware should be applied to

__init__(rate_limit: tuple[DurationUnit, int], *, store: str = 'rate_limit', identifier_for_request: Callable[[Request[Any, Any, Any]], str] = <function get_remote_address>, check_throttle_handler: Callable[[Request[Any, Any, Any]], SyncOrAsyncUnion[bool]] | None = None, set_rate_limit_headers: bool = True, rate_limit_policy_header_key: str = 'RateLimit-Policy', rate_limit_limit_header_key: str = 'RateLimit-Limit', rate_limit_remaining_header_key: str = 'RateLimit-Remaining', rate_limit_reset_header_key: str = 'RateLimit-Reset', exclude: str | list[str] | None = None, exclude_opt_key: str | None = None) → None

Initialize RateLimitMiddleware.

Parameters:

• rate_limit – A tuple containing a time unit (second, minute, hour, day) and quantity, e.g. (“day”, 1) or (“minute”, 5).

• store – Name of the Store to use, looked up on the application’s store registry.

• identifier_for_request – A callable that receives the request and returns an identifier for which the limit should be applied.

• check_throttle_handler – Handler callable that receives the request instance, returning a boolean dictating whether or not the request should be checked for rate limiting.

• set_rate_limit_headers – Boolean dictating whether to set the rate limit headers on the response.

• rate_limit_policy_header_key – Key to use for the rate limit policy header.

• rate_limit_limit_header_key – Key to use for the rate limit limit header.

• rate_limit_remaining_header_key – Key to use for the rate limit remaining header.

• rate_limit_reset_header_key – Key to use for the rate limit reset header.

• exclude – A pattern or list of patterns to skip in the rate limiting middleware, matched against the handler path.

• exclude_opt_key – An identifier to use on routes to disable rate limiting for a particular route.

async handle(scope: Scope, receive: Receive, send: Send, next_app: ASGIApp) → None

Handle ASGI call.

Parameters:

• scope – The ASGI connection scope.

• receive – The ASGI receive function.

• send – The ASGI send function.

• next_app – The next ASGI application in the middleware stack to call.

Returns:

None

create_send_wrapper(send: Send, cache_object: CacheObject) → Send

Create a send function that wraps the original send to inject response headers.

Parameters:

• send – The ASGI send function.

• cache_object – A StorageObject instance.

Returns:

Send wrapper callable.

async retrieve_cached_history(key: str, store: Store) → CacheObject

Retrieve a list of time stamps for the given duration unit.

Parameters:

• key – Cache key.

• store – A Store

Returns:

An CacheObject.

async set_cached_history(key: str, cache_object: CacheObject, store: Store) → None

Store history extended with the current timestamp in cache.

Parameters:

• key – Cache key.

• cache_object – A CacheObject.

• store – A Store

Returns:

None

async should_check_request(request: Request[Any, Any, Any]) → bool

Return a boolean indicating if a request should be checked for rate limiting.

Parameters:

request – A Request instance.

Returns:

Boolean dictating whether the request should be checked for rate-limiting.

create_response_headers(cache_object: CacheObject) → dict[str, str]

Create ratelimit response headers.

Notes

• see the IETF RateLimit draft

Parameters:

cache_object – A CacheObject.

Returns:

A dict of http headers.

litestar.middleware.rate_limit.get_remote_address(request: Request[Any, Any, Any]) → str

Get a client’s remote address from a Request

Parameters:

request – A Request instance.

Returns:

An address, uniquely identifying this client

litestar.middleware.rate_limit.DurationUnit

alias of Literal[‘second’, ‘minute’, ‘hour’, ‘day’]