python fastapi统一捕获请求和响应

服务端代码

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import time
import json
from typing import Dict, Anyapp = FastAPI()@app.middleware("http")
async def log_request_response(request: Request, call_next):# 记录请求开始时间start_time = time.time()# 准备请求日志数据request_info: Dict[str, Any] = {"method": request.method,"url": str(request.url),"headers": dict(request.headers),"query_params": dict(request.query_params),}# 尝试获取请求体（对于非GET请求）if request.method not in ["GET", "HEAD"]:try:body = await request.json()request_info["body"] = bodyexcept:body = await request.body()if body:request_info["body"] = body.decode()# 打印请求信息print("\n" + "="*50)print("Incoming Request:")print(json.dumps(request_info, indent=2, ensure_ascii=False))# 调用路由处理函数response = await call_next(request)# 计算处理时间process_time = time.time() - start_time# 捕获响应体response_body = b""async for chunk in response.body_iterator:response_body += chunk# 准备响应日志数据response_info = {"status_code": response.status_code,"process_time": f"{process_time:.3f}s","body": json.loads(response_body.decode()) if response_body else None,"headers": dict(response.headers),}# 打印响应信息print("\nOutgoing Response:")print(json.dumps(response_info, indent=2, ensure_ascii=False))print("="*50 + "\n")# 重新构造响应（因为我们已经读取了body_iterator）return JSONResponse(content=response_info["body"],status_code=response.status_code,headers=dict(response.headers),)# 测试路由
@app.post("/test")
async def test_endpoint(data: dict):"""测试端点，返回处理后的数据"""return {"message": "Data processed successfully","received_data": data,"additional_info": {"timestamp": time.time()}}@app.get("/hello")
async def hello(name: str = "World"):"""简单的问候端点"""return {"message": f"Hello, {name}!"}if __name__ == "__main__":import uvicornuvicorn.run(app, host="0.0.0.0", port=8000)

客户端测试代码

import requestsdef test_get():url = "http://localhost:8000/hello"params = {"key1": "这是一个get请求测试值","key2": "value2"}# 发送 GET 请求response = requests.get(url, params=params)# 检查响应状态码if response.status_code == 200:# 获取响应内容 (JSON 格式)data = response.json()print("GET 请求成功，响应数据:", data)else:print(f"GET 请求失败，状态码: {response.status_code}")def test_post():url = "http://localhost:8000/test"headers = {"Content-Type": "application/json","Authorization": "Bearer your_token_here"  # 如果需要认证}data = {"key1": "这是一个post请求测试值","key2": "value2"}# 发送 POST 请求response = requests.post(url, headers=headers, json=data)# 检查响应状态码if response.status_code == 200:# 获取响应内容 (JSON 格式)result = response.json()print("POST 请求成功，响应数据:", result)else:print(f"POST 请求失败，状态码: {response.status_code}")print("错误信息:", response.text)test_get()
test_post()