[开源推荐] throttled-py：支持多种策略及存储选项的 Python 限流库

[ZhuoZhuoCrayon]

项目地址

https://github.com/ZhuoZhuoCrayon/throttled-py

类别

Python

项目标题

支持多种策略及存储选项的 Python 限流库

项目描述

• 支持 Redis、内存（线程安全）作为存储后端。
• 支持「固定窗口」「滑动窗口」「令牌桶」「漏桶」「GCRA」作为限流策略。
• 为 Web 服务、IO 密集型提供应对超额流量的保护机制。

亮点

• 提供业内通用的各类限流策略实现。
• 线程、并发安全：Redis（基于 Lua 实现限流算法）、In-Memory（基于 RLock）。
• 策略、存储可配置：提供灵活的限流策略、配额设置 API，文档详尽。
• 支持装饰器模式。

示例代码

1）安装

$ pip install throttled-py

2）通用 API

• limit：消耗请求，返回 RateLimitResult。
• peek：获取指定 Key 的限流器状态，返回 RateLimitState。

from throttled import Throttled

# 参数全部缺省时，默认初始化一个基于「内存」、每分钟允许通过 60 个请求、使用「令牌桶算法」的限流器。
throttle = Throttled()

# 消耗 1 次请求，输出：RateLimitResult(limited=False,
# state=RateLimitState(limit=60, remaining=59, reset_after=1))
print(throttle.limit("key", 1))
# 获取限流器状态，输出：RateLimitState(limit=60, remaining=59, reset_after=1)
print(throttle.peek("key"))

# 消耗 60 次请求，触发限流，输出：RateLimitResult(limited=True,
# # state=RateLimitState(limit=60, remaining=59, reset_after=1))
print(throttle.limit("key", 60))

3）作为装饰器

from throttled import Throttled, rate_limter, exceptions

# 创建一个每秒允许通过 1 次的限流器。
@Throttled(key="/ping", quota=rate_limter.per_min(1))
def ping() -> str:
    return "ping"

ping()

try:
    ping()      # 当触发限流时，抛出 LimitedError。
except exceptions.LimitedError as exc:
    print(exc)  # Rate limit exceeded: remaining=0, reset_after=60
    # 在异常中获取限流结果：RateLimitResult(limited=True, 
    # state=RateLimitState(limit=1, remaining=0, reset_after=60))
    print(exc.rate_limit_result)

4）指定 Redis 作为存储后端

from throttled import RateLimiterType, Throttled, rate_limter, store

@Throttled(
    key="/api/products",
    using=RateLimiterType.TOKEN_BUCKET.value,
    quota=rate_limter.per_min(1),
    # 🌟 使用 Redis 作为存储后端
    store=store.RedisStore(server="redis://127.0.0.1:6379/0", options={"PASSWORD": ""}),
)
def products() -> list:
    return [{"name": "iPhone"}, {"name": "MacBook"}]

products()
# raise LimitedError: Rate limit exceeded: remaining=0, reset_after=60
products()

截图或演示视频