在Python中如何利用Flask实现基于Token的API调用频率限制？-Python教程-PHP中文网

flask-limiter的token_bucket策略易失效，因默认memorystorage在多进程（如gunicorn）下各worker独立维护令牌桶，导致限流完全失效；生产必须改用redisstorage等共享存储，并规范rate格式与key_func。

在python中如何利用flask实现基于token的api调用频率限制？

为什么直接用 `flask-limiter` 的 `token_bucket` 策略容易失效？

因为默认的 token_bucket 实现依赖内存存储（MemoryStorage），在多进程或部署到 Gunicorn/UWSGI 时，每个 worker 维护独立桶，实际限流完全失效。你看到本地测试正常，上线后 QPS 翻倍甚至失控，大概率是这个原因。

实操建议：

SUPAH Token Guardian

支持 21+ 条 EVM 链的盘前代币安全扫描。六层深度检测：合约安全、流动性健康度、部署者画像、持币分布、交易模式等。

下载

必须显式指定共享存储后端，如 Redis：RedisStorage("redis://localhost:6379")
避免用 @limiter.limit("100 per day") 这类模糊表达，改用明确速率格式："100/minute" 或 "5/second"
若无 Redis，可用 MemcachedStorage，但别用 MemoryStorage 上生产

如何按用户 Token（而非 IP）做限流？

Flask-Limiter 默认使用 get_ipaddr 提取 key，要改成从请求头读取 Token 并哈希，否则所有用户共用一个桶。

实操建议：

定义自定义 key_func：

def get_token_key():
    auth = request.headers.get("Authorization")
    if auth and auth.startswith("Bearer "):
        token = auth[7:]
        return hashlib.sha256(token.encode()).hexdigest()[:16]
    return "anonymous"

注册时传入：@limiter.limit("10/minute", key_func=get_token_key)
注意：不要直接返回原始 Token（泄露风险），也不要用完整哈希（key 过长影响 Redis 性能）

`flask-limiter` 的 `key_func` 在什么时机执行？

它在每次请求进入限流装饰器时调用，早于视图函数执行。这意味着你可以安全地访问 request、解析 header、查数据库（但不推荐——会拖慢限流路径）。

常见错误现象：

在 key_func 里调用 db.query.filter(...).first()，导致每个请求都查库，限流反而成性能瓶颈
误以为 key_func 可以异步，实际 Flask-Limiter 不支持 async key_func（会报 RuntimeWarning: coroutine was never awaited）
返回 None 或空字符串，Limiter 会 fallback 到 IP，行为不可控

如何让被限流的请求返回标准 JSON 错误？

默认返回 HTML 页面，API 场景下必须重写响应格式。

实操建议：

注册错误处理器：

@limiter.request_filter
def ip_whitelist():
    return request.remote_addr in ["127.0.0.1"]

@limiter.exempt
def exempt_view():
    pass

@app.errorhandler(429)
def ratelimit_handler(e):
    return {"error": "rate limit exceeded", "retry_after": e.description}, 429