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

如何在 FastAPI 中使用 SlowAPI 为所有路由统一配置全局速率限制

本文介绍如何通过 SlowAPI 的 SlowAPIMiddleware 为 FastAPI 应用设置全局默认速率限制，无需为每个路由手动添加装饰器，并支持按需豁免特定端点或扩展为基于用户角色的动态限流策略。

本文介绍如何通过 slowapi 的 `slowapimiddleware` 为 fastapi 应用设置全局默认速率限制，无需为每个路由手动添加装饰器，并支持按需豁免特定端点或扩展为基于用户角色的动态限流策略。

在构建高可用 API 服务时，合理的速率限制（Rate Limiting）是防止滥用、保障系统稳定性的关键一环。SlowAPI 作为专为 FastAPI 设计的限流库，不仅支持 Redis 等外部后端，也原生提供轻量级的内存（in-memory）限流能力——非常适合中小型项目或开发/测试环境。

最简洁高效的全局限流方式，不是手动编写中间件去调用 limiter 内部逻辑，也不是逐个装饰 endpoint，而是直接启用 SlowAPI 官方推荐的 SlowAPIMiddleware。该中间件会自动拦截所有 HTTP 请求，依据 Limiter 实例中配置的 default_limits 对未显式指定限流规则的路由统一生效。

以下是最小可行示例：

from fastapi import FastAPI, Request, Depends
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

# 初始化 Limiter，指定默认限流规则（如：1 次/分钟）
limiter = Limiter(
    key_func=get_remote_address,
    default_limits=["1/minute"]  # ✅ 全局默认规则：所有路由均受此限制
)

app = FastAPI()
app.state.limiter = limiter  # 必须绑定到 app.state，供中间件访问

# 注册异常处理器，返回标准 429 响应
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
# 启用 SlowAPI 中间件 —— 核心步骤
app.add_middleware(SlowAPIMiddleware)

@app.get("/")
async def home():
    return {"message": "You can call this once per minute"}

@app.get("/public")
@limiter.exempt  # ✅ 豁免特定路由（如公开健康检查接口）
async def public_endpoint():
    return {"status": "ok"}

✅ 关键点说明：

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

下载

default_limits 是全局兜底规则，对所有未显式调用 @limiter.limit(...) 的路由生效；

@limiter.exempt 可用于完全跳过限流（如 /health、/docs），自 SlowAPI v0.1.9+ 起已完整支持 async def 路由，请确保升级至最新版（pip install --upgrade slowapi）；

get_remote_address 作为 key 函数，按客户端 IP 限流；你也可自定义 key（例如结合 JWT 用户 ID 或订阅等级）实现更精细控制。

进阶：基于用户身份的动态限流

若需根据用户权限（如 free/premium）差异化限流，可扩展 key_func 并在路由中注入用户上下文：

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt

# 示例：从 token 解析用户类型
async def get_user_plan(request: Request) -> str:
    auth = HTTPBearer()
    try:
        credentials: HTTPAuthorizationCredentials = await auth(request)
        payload = jwt.decode(credentials.credentials, "secret", algorithms=["HS256"])
        return payload.get("plan", "free")  # e.g., "free" or "premium"
    except Exception:
        return "free"

# 自定义限流键：组合 IP + 订阅等级
def dynamic_key(request: Request):
    ip = get_remote_address(request)
    plan = request.state.plan if hasattr(request.state, "plan") else "free"
    return f"{ip}:{plan}"

# 在中间件中提前解析 plan 并存入 request.state（需配合依赖注入）
@app.middleware("http")
async def add_plan_to_request(request: Request, call_next):
    request.state.plan = await get_user_plan(request)
    return await call_next(request)

# 重新初始化 limiter，使用动态 key
limiter = Limiter(
    key_func=dynamic_key,
    default_limits={
        "free": "10/minute",
        "premium": "100/minute"
    }.__getitem__  # ⚠️ 注意：此处需配合自定义逻辑，实际建议用装饰器覆盖
)

但更推荐的做法是：保留全局默认值，对 premium 用户路由显式覆盖：

@app.get("/premium/data")
@limiter.limit("100/minute", key_func=get_remote_address)
async def premium_data():
    return {"data": "high-speed content"}