接上篇：一文掌握异步web框架FastAPI（四）-CSDN博客
目录
七、中间件
15、测试环境中间件
16、访问速率限制中间件，即限制每个IP特定时间内的请求数（基于内存，生产上要使用数据库）
1）限制单ip访问速率
2）增加限制单ip并发（跟上面的一样，也是限制每个IP特定时间内的请求数，另一种写法）
3）增加封禁IP
17. 请求体解析中间件
18. 自定义认证中间件
19. 重试机制中间件
20. 请求频率统计中间件（各个接口的访问次数）
21. 路径重写中间件
七、中间件
15、测试环境中间件
这个中间件用于识别请求是否来自于测试环境，并采取相应的措施，如禁用缓存或跳过某些安全检查。
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import os
import logging# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI()@app.middleware("http")
async def test_environment_middleware(request: Request, call_next):try:# 检查环境变量test_environment_enabled = os.getenv("TEST_ENVIRONMENT_ENABLED", "false").lower() == "true"if "X-Test-Environment" in request.headers and request.headers["X-Test-Environment"].lower() == "true":if test_environment_enabled:logger.info("Test environment header detected. Enabling test environment specific logic.")# 测试环境特有的处理# 例如：设置数据库连接为测试数据库、开启调试模式等passelse:logger.warning("Test environment header detected, but TEST_ENVIRONMENT_ENABLED is not set to true.")return JSONResponse({"message": "Test environment not enabled"}, status_code=403)response = await call_next(request)return responseexcept Exception as e:logger.error(f"Error in test_environment_middleware: {str(e)}")return JSONResponse({"message": "Internal server error"}, status_code=500)@app.get("/")
async def root():return {"message": "Hello World"}if __name__ == "__main__":import uvicornuvicorn.run(app, host="0.0.0.0", port=8000)
请求：
import requests# FastAPI应用的URL
url = "http://127.0.0.1:8000/"
# 发送GET请求到根路径，并包含测试环境头部
response = requests.get(url, headers={"X-Test-Environment": "true"})
# 打印响应状态码和内容
print(f"Status Code: {response.status_code}")
print(f"Response Content: {response.json()}")
本地未启用测试环境：
16、访问速率限制中间件，即限制每个IP特定时间内的请求数（基于内存，生产上要使用数据库）
1）限制单ip访问速率
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import time
from collections import defaultdict
import loggingapp = FastAPI()
# 初始化rate_limits字典
rate_limits = defaultdict(list)
# 配置日志
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)# 定义中间件
@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):ip = request.headers.get("X-Forwarded-For", request.client.host).split(",")[0].strip()now = time.time()# 移除旧的请求时间戳rate_limits[ip] = [t for t in rate_limits[ip] if t > now - 60]# 检查速率限制if len(rate_limits[ip]) >= 10: # 限制为每分钟10个请求logger.debug(f"Rate limit exceeded for IP: {ip}")return JSONResponse(status_code=429,content={"detail": "Too Many Requests"})# 添加当前请求的时间戳rate_limits[ip].append(now)# 继续处理请求response = await call_next(request)return response# 定义路由
@app.get("/")
async def root():return {"message": "Hello World"}# 启动应用
if __name__ == "__main__":import uvicornuvicorn.run(app, host="0.0.0.0", port=8000, log_level="debug")
请求：
import requests
import time# FastAPI应用的URL
url = "http://127.0.0.1:8000/"
# 用于测试的IP地址
ip = "127.0.0.1"
# 设置请求头，模拟X-Forwarded-For
headers = {"X-Forwarded-For": ip
}# 发送多个请求以测试速率限制
def test_rate_limit():for i in range(15):try:response = requests.get(url, headers=headers)print(f"Request {i + 1}: Status Code: {response.status_code}")if response.status_code == 429:print(f"Request {i + 1}: {response.json()}")except Exception as e:print(f"Request {i + 1}: Error: {e}")time.sleep(1) # 等待1秒
)
