taolib.testing.rate_limiter.store

Rate limit storage backends.

Provides Protocol definition, Redis implementation, and in-memory implementation for testing.

Classes

RateLimitStoreProtocol	限流存储接口。
RedisRateLimitStore	基于 Redis Sorted Set 的滑动窗口实现。
InMemoryRateLimitStore	内存版限流存储，用于测试。

Module Contents

class taolib.testing.rate_limiter.store.RateLimitStoreProtocol

Bases: Protocol

限流存储接口。

async record_request(identifier: str, path: str, method: str, timestamp: float, window_seconds: int) → int

记录请求并返回当前窗口内请求数。

参数:

• identifier -- 用户标识符

• path -- 请求路径

• method -- HTTP 方法

• timestamp -- 请求时间戳

• window_seconds -- 窗口大小

返回:

当前窗口内请求数

async get_request_count(identifier: str, path: str, method: str, window_seconds: int) → int

获取当前窗口内请求数。

参数:

• identifier -- 用户标识符

• path -- 请求路径

• method -- HTTP 方法

• window_seconds -- 窗口大小

返回:

请求数

async get_oldest_in_window(identifier: str, path: str, method: str, window_seconds: int) → float | None

获取窗口内最早的请求时间戳。

参数:

• identifier -- 用户标识符

• path -- 请求路径

• method -- HTTP 方法

• window_seconds -- 窗口大小

返回:

最早请求的时间戳，如果窗口为空则返回 None

async increment_stats(identifier: str, path: str) → None

增加统计计数。

参数:

• identifier -- 用户标识符

• path -- 请求路径

async get_top_users(limit: int = 20) → list[tuple[str, int]]

获取请求量最大的用户列表。

参数:

limit -- 返回数量限制

返回:

[(identifier, count), ...] 列表

async get_realtime_requests(window_seconds: int = 60) → dict[str, Any]

获取实时请求统计。

参数:

window_seconds -- 统计窗口

返回:

实时统计数据

class taolib.testing.rate_limiter.store.RedisRateLimitStore(redis_client: Any)

基于 Redis Sorted Set 的滑动窗口实现。

使用 Sorted Set 存储请求时间戳： - Score: 请求时间戳 - Member: "{timestamp}:{unique_id}" 保证唯一性

参数:

redis_client -- Redis 异步客户端（redis.asyncio.Redis）

_redis

async record_request(identifier: str, path: str, method: str, timestamp: float, window_seconds: int) → int

async get_request_count(identifier: str, path: str, method: str, window_seconds: int) → int

async get_oldest_in_window(identifier: str, path: str, method: str, window_seconds: int) → float | None

async increment_stats(identifier: str, path: str) → None

async get_top_users(limit: int = 20) → list[tuple[str, int]]

async get_realtime_requests(window_seconds: int = 60) → dict[str, Any]

class taolib.testing.rate_limiter.store.InMemoryRateLimitStore

内存版限流存储，用于测试。

使用字典模拟 Sorted Set 行为。

_windows: dict[str, list[float]]

_stats: dict[str, int]

_realtime: list[tuple[float, str]] = []

async record_request(identifier: str, path: str, method: str, timestamp: float, window_seconds: int) → int

async get_request_count(identifier: str, path: str, method: str, window_seconds: int) → int

async get_oldest_in_window(identifier: str, path: str, method: str, window_seconds: int) → float | None

async increment_stats(identifier: str, path: str) → None

async get_top_users(limit: int = 20) → list[tuple[str, int]]

async get_realtime_requests(window_seconds: int = 60) → dict[str, Any]