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
<h1>简单内存存储：{ip: [timestamp1, timestamp2, ...]}</h1><p>ip_request_times: Dict[str, List[float]] = {}</p><h1>限流配置</h1><p>MAX_REQUESTS = 10
TIME_WINDOW_SEC = 60</p><p>def rate_limit_dependency(request: Request):
client_ip = request.client.host
now = time.time()</p><pre class='brush:python;toolbar:false;'># 清理过期时间戳
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 保护共享字典操作：
- 为每个 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 会导致限流完全失效。
