本文介绍如何通过 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"}注意事项与最佳实践
- ❗
SlowAPIMiddleware依赖app.state.limiter，务必在添加中间件前完成赋值； - ? 不要自行实现“手动中间件 + 调用 limiter.check()”，这会绕过 SlowAPI 的异常处理与响应格式化，且易出竞态问题；
- ? 内存限流不适用于多进程部署（如
uvicorn --workers 4），此时必须使用 Redis 后端； - ? 生产环境建议搭配日志记录（如
logger.info(f"Rate limit hit for {key}")）和监控告警。
通过以上方式，你既能以声明式、低侵入的方式实现全站速率防护，又能灵活适配业务增长带来的分级限流需求。
