taolib.testing.rate_limiter.limiter

taolib.testing.rate_limiter.limiter#

Rate Limiter core engine.

Implements the sliding window rate limiting logic with whitelist support and path-specific rules.

Classes#

RateLimiter

限流引擎核心类。

Module Contents#

class taolib.testing.rate_limiter.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 方法