rate_limit

class litestar.middleware.rate_limit.CacheObject

基类：object

缓存对象元数据的表示。

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

class litestar.middleware.rate_limit.RateLimitConfig

基类：object

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.

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.

RateLimitMiddleware 的别名

set_rate_limit_headers: bool = True

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

rate_limit_policy_header_key: str = 'RateLimit-Policy'

Key to use for the rate limit policy header.

__init__(rate_limit: tuple[DurationUnit, int], exclude: str | list[str] | None = None, exclude_opt_key: str | None = None, 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_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: DefineMiddleware

使用此属性将配置插入到应用层之一的中间件列表中。 示例： .. code-block:: python from litestar import Litestar, Request, get from litestar.middleware.rate_limit import RateLimitConfig # 限制为每分钟 10 次请求，排除 schema 路径 throttle_config = RateLimitConfig(rate_limit=("minute", 10), exclude=["/schema"]) @get("/") def my_handler(request: Request) -> None: ... app = Litestar(route_handlers=[my_handler], middleware=[throttle_config.middleware]) Returns: 一个 DefineMiddleware 实例，其中 self 作为配置关键字参数值包含在内。

get_store_from_app(app: Litestar) → Store

从 Litestar 实例中获取在 store 中定义的存储。

class litestar.middleware.rate_limit.RateLimitMiddleware

基类：AbstractMiddleware

速率限制中间件。

__init__(app: ASGIApp, config: RateLimitConfig) → None

初始化 RateLimitMiddleware。 Args: app: 要调用的下一个 next ASGI 应用。 config: RateLimitConfig 的实例。

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

创建一个 send 函数，用于包装原始的 send 以注入响应头。 Args: send: ASGI send 函数。 cache_object: 一个 StorageObject 实例。 Returns: send 包装后的可调用对象。

cache_key_from_request(request: Request[Any, Any, Any]) → str

Get a cache-key from a Request

参数:

request -- A Request instance.

返回:

A cache key.

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

获取给定持续时间单位的时戳列表。 Args: key: 缓存键。 store: 一个 Store 实例。 Returns: 一个 CacheObject 实例。

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

在缓存中存储带有当前时间戳的历史记录。 Args: key: 缓存键。 cache_object: 一个 CacheObject。 store: 一个 Store Returns: None

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

返回一个布尔值，指示是否应对请求进行速率限制检查。 Args: request: 一个 Request 实例。 Returns: 布尔值，规定是否应对请求进行速率限制检查。

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

创建速率限制响应头。 Notes: * 参见 IETF RateLimit draft Args: cache_object: 一个 CacheObject。 Returns: 一个包含 HTTP 头的字典。

litestar.middleware.rate_limit.DurationUnit

Literal['second', 'minute', 'hour', 'day'] 的别名