taolib.testing.rate_limiter.middleware

FastAPI middleware for rate limiting.

Classes

RateLimitMiddleware

FastAPI 限流中间件。

Functions

extract_client_ip(→ str)	提取客户端 IP 地址。
extract_user_id(→ str | None)	尝试从请求中提取用户 ID。

Module Contents

taolib.testing.rate_limiter.middleware.extract_client_ip(request: fastapi.Request) → str

提取客户端 IP 地址。

优先级： 1. X-Forwarded-For 第一个 IP（代理场景） 2. X-Real-IP 3. request.client.host（直接连接）

参数:

request -- FastAPI 请求对象

返回:

客户端 IP 地址

taolib.testing.rate_limiter.middleware.extract_user_id(request: fastapi.Request) → str | None

尝试从请求中提取用户 ID。

从 JWT token 的 sub 字段或自定义头 X-User-ID 提取。

参数:

request -- FastAPI 请求对象

返回:

用户 ID 或 None

class taolib.testing.rate_limiter.middleware.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 -- 下一个处理函数

返回:

响应对象