taolib.testing.rate_limiter

taolib.testing.rate_limiter#

Rate Limiter Module

滑动窗口限流中间件，基于 Redis 实现，为 FastAPI 应用提供： - 按用户/IP 差异化限流 - 接口级限流配置 - 白名单机制 - 违规监控与统计

Usage:

from taolib.testing.rate_limiter import RateLimiter, RateLimitMiddleware

Submodules#

Exceptions#

RateLimitExceededError

限流阈值超出异常。

Classes#

RateLimitConfig

限流配置。

RateLimiter

限流引擎核心类。

RateLimitMiddleware

FastAPI 限流中间件。

Package Contents#

class taolib.testing.rate_limiter.RateLimitConfig#

Bases: pydantic.BaseModel

限流配置。

enabled: bool#
default_limit: int#
window_seconds: int#
path_rules: dict[str, PathRule]#
whitelist: WhitelistConfig#
redis_url: str#
mongo_violation_ttl_days: int#
mongo_collection: str#
classmethod from_toml(config_path: str) → RateLimitConfig#

从 TOML 文件加载配置。

参数:

config_path -- TOML 配置文件路径

返回:

限流配置实例

exception taolib.testing.rate_limiter.RateLimitExceededError(*, limit: int, window_seconds: int, retry_after: int, identifier: str)#

Bases: Exception

限流阈值超出异常。

当请求频率超过配置的限流阈值时抛出。

limit#

限流阈值

window_seconds#

滑动窗口大小（秒）

retry_after#

建议重试等待时间（秒）

identifier#

被限流的标识符（用户ID或IP）

reset_timestamp#

窗口重置时间戳

limit#
window_seconds#
retry_after#
identifier#
reset_timestamp#
class taolib.testing.rate_limiter.RateLimiter(config: taolib.testing.rate_limiter.config.RateLimitConfig, store: taolib.testing.rate_limiter.store.RateLimitStoreProtocol)#

限流引擎核心类。

负责： - 白名单检查 - 路径规则匹配 - 滑动窗口计数 - 限流决策

参数:

  • config -- 限流配置

  • store -- 存储后端（Redis 或内存）

_config#
_store#
is_whitelisted_ip(ip: str) → bool#

检查 IP 是否在白名单中。

参数:

ip -- IP 地址

返回:

是否在白名单中

is_whitelisted_user(user_id: str) → bool#

检查用户 ID 是否在白名单中。

参数:

user_id -- 用户 ID

返回:

是否在白名单中

is_bypass_path(path: str) → bool#

检查路径是否 bypass 限流。

参数:

path -- 请求路径

返回:

是否 bypass

_get_rule_for_path(path: str, method: str) → tuple[int, int]#

获取路径对应的限流规则。

参数:

  • path -- 请求路径

  • method -- HTTP 方法

返回:

(limit, window_seconds) 元组

async check_limit(identifier: str, path: str, method: str = 'GET') → taolib.testing.rate_limiter.models.RateLimitResult#

检查请求是否超出限流阈值。

参数:

  • identifier -- 用户标识符（如 "user:abc123" 或 "ip:192.168.1.1"）

  • path -- 请求路径

  • method -- HTTP 方法

返回:

限流检查结果

抛出:

RateLimitExceededError -- 超出限流阈值时

async record_request(identifier: str, path: str, method: str = 'GET') → None#

记录请求到存储。

参数:

  • identifier -- 用户标识符

  • path -- 请求路径

  • method -- HTTP 方法

class taolib.testing.rate_limiter.RateLimitMiddleware(app: Any, limiter: taolib.testing.rate_limiter.limiter.RateLimiter, violation_tracker: taolib.testing.rate_limiter.violation_tracker.ViolationTracker | None = None, logger: logging.Logger | None = None)#

Bases: starlette.middleware.base.BaseHTTPMiddleware

FastAPI 限流中间件。

功能： - 提取用户标识符（用户 ID 或 IP） - 检查白名单和 bypass 路径 - 调用限流引擎检查请求 - 注入限流响应头 - 记录违规行为

参数:

  • app -- ASGI 应用

  • limiter -- 限流引擎实例

  • violation_tracker -- 违规追踪器实例（可选）

  • logger -- 日志记录器

_limiter#
_violation_tracker = None#
_logger#
async dispatch(request: fastapi.Request, call_next: Any) → fastapi.Response#

处理请求。

参数:

  • request -- 接收的请求

  • call_next -- 下一个处理函数

返回:

响应对象