如何在新版Sanic中实现API限流并返回剩余请求数？

问题描述 #

我正尝试在Sanic中实现请求限流功能，但PyPI上的sanic-limiter包已多年未维护，无法适配当前Sanic版本。找到该包的一个GitHub分叉版本，可通过以下代码设置全局请求限制：

limiter = Limiter(
        app,
        global_limits = [
            "1 per second",
            "500 per day"
            ],
        key_func = get_remote_address
    )

但该版本的其他示例无法正常运行，推测其与当前Sanic版本仍不兼容。

sanic-limiter是Flask-Limiter的Sanic适配版本，Flask-Limiter可通过after_request钩子返回限流相关信息：

@app.after_request
def add_headers(response):
    if limiter.current_limit:
        response.headers["RemainingLimit"] = limiter.current_limit.remaining
        response.headers["ResetAt"] = limiter.current_limit.reset_at
        response.headers["MaxRequests"] = limiter.current_limit.limit.amount
        response.headers["WindowSize"] = limiter.current_limit.limit.get_expiry()
        response.headers["Breached"] = limiter.current_limit.breached
    return response

我尝试将其转换为Sanic的响应中间件：

@app.middleware("response")
async def add_headers(request, response):
    if limiter.current_limit:
        response.headers["RemainingIDs"] = str(
            limiter.current_limit.remaining
            )
        
        response.headers["ResetAt"] = str(
            limiter.current_limit.reset_at
            )
    
    return response

但发现该limiter并不具备这些属性。想知道在新版Sanic中实现限流并返回剩余请求数的替代方案或解决办法。

可行解决方案 #

方案1：基于limits库手动实现限流 #

limits是一个通用的限流库，支持多种存储后端（内存、Redis等），可以直接在Sanic中集成：

1. 安装依赖：

pip install limits

2. 实现限流逻辑和响应头注入：

from sanic import Sanic, Request, HTTPResponse
from limits import parse
from limits.storage import MemoryStorage
from limits.strategies import FixedWindowRateLimiter

app = Sanic("RateLimiterApp")

# 初始化限流器，支持全局或路由级配置
storage = MemoryStorage()
limiter = FixedWindowRateLimiter(storage)
# 全局限流规则：1次/秒，500次/天
global_limits = [parse("1/second"), parse("500/day")]

@app.middleware("request")
async def check_rate_limit(request: Request):
    client_ip = request.remote_addr
    # 检查所有全局限流规则
    for limit in global_limits:
        if not limiter.hit(limit, client_ip):
            # 获取限流剩余信息
            remaining = limiter.get_remaining(limit, client_ip)
            reset_at = limiter.get_reset(limit, client_ip)
            # 构建限流响应
            response = HTTPResponse(
                body="Too Many Requests",
                status=429,
                headers={
                    "RemainingLimit": str(remaining),
                    "ResetAt": str(reset_at),
                    "MaxRequests": str(limit.amount),
                    "WindowSize": str(limit.get_expiry())
                }
            )
            return response
    # 将限流信息存入request上下文，供响应中间件使用
    request.ctx.limit_info = {}
    for limit in global_limits:
        request.ctx.limit_info[str(limit)] = {
            "remaining": limiter.get_remaining(limit, client_ip),
            "reset_at": limiter.get_reset(limit, client_ip),
            "max_requests": limit.amount,
            "window_size": limit.get_expiry()
        }

@app.middleware("response")
async def add_limit_headers(request: Request, response: HTTPResponse):
    if hasattr(request.ctx, "limit_info"):
        # 示例：取第一个规则的信息返回，可根据需求调整
        first_limit_info = next(iter(request.ctx.limit_info.values()))
        response.headers["RemainingLimit"] = str(first_limit_info["remaining"])
        response.headers["ResetAt"] = str(first_limit_info["reset_at"])
        response.headers["MaxRequests"] = str(first_limit_info["max_requests"])
        response.headers["WindowSize"] = str(first_limit_info["window_size"])

@app.get("/")
async def hello(request):
    return {"message": "Hello World"}

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

方案2：封装路由级限流装饰器 #

如果需要针对不同路由设置差异化限流规则，可以将逻辑封装为装饰器：

def rate_limit(limit_str):
    def decorator(func):
        async def wrapped(request: Request, *args, **kwargs):
            client_ip = request.remote_addr
            limit = parse(limit_str)
            if not limiter.hit(limit, client_ip):
                remaining = limiter.get_remaining(limit, client_ip)
                reset_at = limiter.get_reset(limit, client_ip)
                return HTTPResponse(
                    body="Too Many Requests",
                    status=429,
                    headers={
                        "RemainingLimit": str(remaining),
                        "ResetAt": str(reset_at),
                        "MaxRequests": str(limit.amount)
                    }
                )
            # 存入限流信息供响应中间件使用
            request.ctx.limit_info = {
                "remaining": limiter.get_remaining(limit, client_ip),
                "reset_at": limiter.get_reset(limit, client_ip),
                "max_requests": limit.amount
            }
            return await func(request, *args, **kwargs)
        return wrapped
    return decorator

# 使用示例
@app.get("/api/private")
@rate_limit("5 per minute")
async def private_api(request):
    return {"data": "private content"}

关键注意点 #

• 异步支持：分布式场景下，建议使用limits的异步Redis存储后端（limits.storage.async_redis.AsyncRedisStorage），避免阻塞Sanic事件循环
• 响应头规范：配合429状态码时，建议同时返回Retry-After头，提示客户端可重试的时间
• 存储选型：生产环境不建议使用内存存储，避免重启服务后限流计数丢失

内容的提问来源于stack exchange，提问作者Manuel Popp