如何在 FastAPI 中使用 SlowAPI 统一配置全局速率限制-Python教程-PHP中文网

如何在 FastAPI 中使用 SlowAPI 统一配置全局速率限制

本文介绍如何借助 SlowAPI 的 SlowAPIMiddleware 为 FastAPI 所有路由自动应用统一的默认速率限制，无需逐个装饰器手动添加，并支持按需豁免特定接口及未来扩展（如基于用户角色的差异化限流）。

本文介绍如何借助 slowapi 的 `slowapimiddleware` 为 fastapi 所有路由自动应用统一的默认速率限制，无需逐个装饰器手动添加，并支持按需豁免特定接口及未来扩展（如基于用户角色的差异化限流）。

在构建高可用 API 服务时，防止滥用和保障系统稳定性至关重要。SlowAPI 是专为 FastAPI 设计的轻量级限流库，支持内存存储（无需 Redis），开箱即用。与其为每个 @app.get()、@app.post() 等路由重复添加 @limiter.limit("5/minute")，更优雅的方式是声明式地设置全局默认限流策略——这正是 SlowAPIMiddleware 的核心价值。

✅ 正确做法：启用全局中间件 + 默认限流

只需三步即可完成全局限流配置：

初始化 Limiter 并指定 default_limits（如 "1/minute"）；
将 limiter 挂载到 app.state，供中间件识别；
注册 SlowAPIMiddleware 和 RateLimitExceeded 异常处理器。

完整示例代码如下：

from fastapi import FastAPI, Request, Response
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.middleware import SlowAPIMiddleware
from slowapi.errors import RateLimitExceeded

# 1. 创建限流器，设置全局默认规则（所有路由均受此约束）
limiter = Limiter(
    key_func=get_remote_address,  # 按客户端 IP 限流
    default_limits=["1/minute"]     # 每分钟最多 1 次请求
)

app = FastAPI(title="Global Rate Limited API")

# 2. 必须挂载到 app.state.limiter，否则中间件无法工作
app.state.limiter = limiter

# 3. 注册异常处理器与中间件
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
app.add_middleware(SlowAPIMiddleware)

# 示例路由 —— 自动受全局限流保护
@app.get("/")
async def home():
    return {"message": "You're rate-limited to 1 request per minute."}

@app.get("/health")
async def health():
    return {"status": "ok"}

✅ 此配置下，所有未显式豁免的路由均自动应用 1/minute 限流，无需任何装饰器。

? 如何豁免特定路由？

若某些接口（如健康检查 /health 或公开静态资源）不应受限，可使用 @limiter.exempt 装饰器：

FastAPI Flask Proxy

FastAPI + Flask 混合部署最佳实践，解决路由定义、API 代理等常见问题，适用于同时运行 FastAPI API 与 Flask 前端的场景。

下载

@app.get("/unlimited")
@limiter.exempt  # ✅ 完全跳过限流检查
async def unlimited_endpoint():
    return {"message": "No limits applied here."}

⚠️ 注意：确保使用 SlowAPI ≥ 0.1.9（v0.1.9+ 已修复 async def 支持问题）。旧版本中该装饰器对 async def 路由可能失效。

? 进阶：按用户身份动态限流（如免费/付费）

虽然 default_limits 是静态的，但 SlowAPI 支持基于请求上下文的动态键函数。你可以扩展 key_func 实现差异化限流，例如结合 request.state.user 或 JWT 解析结果：

def dynamic_key_func(request: Request) -> str:
    # 假设已通过中间件将用户信息注入 request.state.user
    user = getattr(request.state, "user", None)
    plan = user.plan if user else "free"
    ip = get_remote_address(request)
    return f"{ip}:{plan}"  # 不同 plan 使用独立计数器

limiter = Limiter(
    key_func=dynamic_key_func,
    default_limits=["10/minute"],  # 免费用户默认
    application_limits=["100/minute"]  # 可选：应用级兜底（需手动调用 limiter.limit）
)

再配合自定义依赖项解析用户权限，即可实现「免费用户 10 次/分钟，付费用户 100 次/分钟」的灵活策略。