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 保护共享字典操作：
- 为每个 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 会导致限流完全失效。
