这是你可以做到的（灵感来自this）。一定要检查响应的Content-Type（如下图），这样你就可以通过添加metadata来修改它，只有它是application/json类型。
更新 1
对于要呈现的 OpenAPI (Swagger UI)（/docs 和 /redoc），请确保检查响应中是否不存在 openapi 键，以便仅在这种情况下才能继续修改响应。如果您的响应数据中碰巧有一个具有此类名称的密钥，那么您可以使用 OpenAPI 响应中存在的其他密钥进行额外检查，例如，info、version、paths 和，如果需要，您也可以检查它们的值。
from fastapi import FastAPI, Request, Response
import json
app = FastAPI()
@app.middleware("http")
async def add_metadata_to_response_payload(request: Request, call_next):
response = await call_next(request)
content_type = response.headers.get('Content-Type')
if content_type == "application/json":
response_body = [section async for section in response.body_iterator]
resp_str = response_body[0].decode() # converts "response_body" bytes into string
resp_dict = json.loads(resp_str) # converts resp_str into dict
#print(resp_dict)
if "openapi" not in resp_dict:
data = {}
data["data"] = resp_dict # adds the "resp_dict" to the "data" dictionary
data["metadata"] = {
"some_data_key_1": "some_data_value_1",
"some_data_key_2": "some_data_value_2",
"some_data_key_3": "some_data_value_3"}
resp_str = json.dumps(data, indent=2) # converts dict into JSON string
return Response(content=resp_str, status_code=response.status_code, media_type=response.media_type)
return response
@app.get("/")
async def foo(request: Request):
return {"hello": "world!"}
更新 2
或者，一种可能更好的方法是在中间件函数的开头检查请求的 url 路径（根据您希望将元数据添加到其响应中的路径/路由的预定义列表），并相应地继续。或者，您可以使用Custom APIRoute class in a router。
routes_with_middleware = ["/"]
@app.middleware("http")
async def add_metadata_to_response_payload(request: Request, call_next):
response = await call_next(request)
if request.url.path not in routes_with_middleware:
return response
else:
content_type = response.headers.get('Content-Type')
if content_type == "application/json":
response_body = [section async for section in response.body_iterator]
resp_str = response_body[0].decode() # converts "response_body" bytes into string
resp_dict = json.loads(resp_str) # converts resp_str into dict
data = {}
data["data"] = resp_dict # adds "resp_dict" to the "data" dictionary
data["metadata"] = {
"some_data_key_1": "some_data_value_1",
"some_data_key_2": "some_data_value_2",
"some_data_key_3": "some_data_value_3"}
resp_str = json.dumps(data, indent=2) # converts dict into JSON string
return Response(content=resp_str, status_code=response.status_code, media_type="application/json")
return response
工作示例
from fastapi import FastAPI, Request, Response, Query
from pydantic import constr
from fastapi.responses import JSONResponse
import re
import uvicorn
import json
app = FastAPI()
routes_with_middleware = ["/"]
rx = re.compile(r'^(/items/\d+|/courses/[a-zA-Z0-9]+)$') # support routes with path parameters
my_constr = constr(regex="^[a-zA-Z0-9]+$")
@app.middleware("http")
async def add_metadata_to_response_payload(request: Request, call_next):
response = await call_next(request)
if request.url.path not in routes_with_middleware and not rx.match(request.url.path):
return response
else:
content_type = response.headers.get('Content-Type')
if content_type == "application/json":
response_body = [section async for section in response.body_iterator]
resp_str = response_body[0].decode() # converts "response_body" bytes into string
resp_dict = json.loads(resp_str) # converts resp_str into dict
data = {}
data["data"] = resp_dict # adds "resp_dict" to the "data" dictionary
data["metadata"] = {
"some_data_key_1": "some_data_value_1",
"some_data_key_2": "some_data_value_2",
"some_data_key_3": "some_data_value_3"}
resp_str = json.dumps(data, indent=2) # converts dict into JSON string
return Response(content=resp_str, status_code=response.status_code, media_type="application/json")
return response
@app.get("/")
async def root():
return {"hello": "world!"}
@app.get("/items/{id}")
async def get_item(id: int):
return {"Item": id}
@app.get("/courses/{code}")
async def get_course(code: my_constr):
return {"course_code": code, "course_title": "Deep Learning"}