FastAPI需借助依赖项与缓存实现IP限流:简易版用内存字典记录IP时间戳并清理过期项;增强版加asyncio.Lock防并发冲突;生产环境应换为Redis的sorted set,配合可信IP校验确保安全。

FastAPI 本身不内置 IP 频率限制功能,但可通过中间件或依赖项 + 缓存(如内存缓存或 Redis)轻松实现。下面介绍一种轻量、可直接运行的基于内存的 IP 限流方案,适合开发或低流量场景;也附上扩展到 Redis 的关键思路。
使用 FastAPI 依赖项 + 内存缓存(简易版)
核心思路:提取客户端 IP,按 IP 统计请求次数,设定时间窗口(如 60 秒内最多 10 次),超限则返回 429。
注意:客户端 IP 需从 request.client.host 获取,但在反向代理(如 Nginx)后需改用 X-Forwarded-For 头,本例默认直连;生产环境请务必校验可信代理头。
示例代码:
from fastapi import Depends, FastAPI, Request, HTTPException from starlette.status import HTTP_429_TOO_MANY_REQUESTS from typing import Dict, List, Optional import time简单内存存储:{ip: [timestamp1, timestamp2, ...]}
ip_request_times: Dict[str, List[float]] = {}
限流配置
MAX_REQUESTS = 10 TIME_WINDOW_SEC = 60
def rate_limit_dependency(request: Request): client_ip = request.client.host now = time.time()
# 清理过期时间戳 if client_ip in ip_request_times: ip_request_times[client_ip] = [ t for t in ip_request_times[client_ip] if now - t < TIME_WINDOW_SEC ] else: ip_request_times[client_ip] = [] # 检查是否超限 if len(ip_request_times[client_ip]) >= MAX_REQUESTS: raise HTTPException( status_code=HTTP_429_TOO_MANY_REQUESTS, detail="Too many requests from this IP" ) # 记录本次请求时间 ip_request_times[client_ip].append(now) return Trueapp = FastAPI()
@app.get("/public") def public_endpoint(rate_limited: bool = Depends(rate_limit_dependency)): return {"message": "OK"}
用 asyncio.Lock 避免并发写冲突(推荐增强版)
上面的内存方案在多进程/多线程下不安全,且高并发时可能因竞态导致误放行。FastAPI 常运行于异步环境,可用 asyncio.Lock 保护共享字典操作:
系统功能介绍 1 包含企业网站所必备的功能:企业信息、产品管理、人才招聘、新闻资讯、企业图片、以及视频下载等模块2 由于是从CMS系统的基础上开发而成,因此相对于一些其他的企业网站管理系统,本系统具备更强的可扩展能力,可以胜任从小型工作室到大中型企业网上门户等各种不同规模网站的需求。3 后台管理与模板完全分离,并具备非常灵活的标签技术,可以实现无限制个性化的界面定制4 操作简单,利用已经制作好的模
- 为每个 IP 分配独立锁(避免全局锁瓶颈)
- 用
defaultdict+Lock管理时间戳列表 - 清理和插入逻辑加锁,确保原子性
实际项目中建议封装成可复用的依赖类,支持不同路由配置不同限流规则(如 /login 更严格)。
生产环境:替换为 Redis(推荐)
内存方案无法跨进程/实例共享,上线必须用 Redis。核心逻辑不变,只是把 ip_request_times 换成 Redis 的 sorted set 或 key+list:
- 用
ZREMRANGEBYSCORE清理过期时间戳 - 用
ZCARD判断当前请求数 - 用
ZADD插入新时间戳(score = 时间戳) - Key 格式建议:
flood:{ip}:{endpoint},便于按端点隔离
可借助 aioredis 或 redis-py(v4.2+ 支持 async)实现异步操作,避免阻塞事件循环。
补充:获取真实客户端 IP 的健壮写法
若部署在 Nginx、Cloudflare 后,request.client.host 是代理 IP。应优先读取 X-Forwarded-For,但需验证来源可信(只信任你自己的反向代理):
- 设置
trusted_hosts=["127.0.0.1", "10.0.0.0/8"] - 用
request.headers.get("X-Forwarded-For", "").split(",")[0].strip()取最左 IP - 务必配合
Starlette's TrustedHostMiddleware或手动校验,防止伪造
不加校验直接信任 X-Forwarded-For 会导致限流完全失效。









