在FastAPI应用程序的中间件级别修改响应后，如何在swagger中为所有端点更新响应模式

Question

因此，我为我的大型FastAPI应用程序创建了一个自定义中间件，它以这种方式更改来自所有端点的响应：

所有API的响应模型都不同。然而，我的MDW以统一的方式将元数据添加到所有这些响应中。这就是最终响应对象的样子：

{
"data": <ANY RESPONSE MODEL THAT ALL THOSE ENDPOINTS ARE SENDING>,
"meta_data": 
{
"meta_data_1": "meta_value_1",
"meta_data_2": "meta_value_2",
"meta_data_3": "meta_value_3",
}
}

因此，本质上，所有原始响应都封装在data字段中，meta_data的新字段添加了所有meta_data。这个meta_data模型是统一的，它将永远是这种类型：

"meta_data": 
{
"meta_data_1": "meta_value_1",
"meta_data_2": "meta_value_2",
"meta_data_3": "meta_value_3",
}

现在的问题是，当swagger加载时，它在模式中显示原始响应模型，而不是已经准备好的最终响应模型。如何改变招摇以正确反映这一点？我试过这个：

# This model is common to all endpoints!
# Since we are going to add this for all responses 
class MetaDataModel(BaseModel):
meta_data_1: str
meta_data_2: str
meta_data_3: str
class FinalResponseForEndPoint1(BaseModel):
data: OriginalResponseForEndpoint1
meta_data: MetaDataModel
class FinalResponseForEndPoint2(BaseModel):
data: OriginalResponseForEndpoint2
meta_data: MetaDataModel
and so on ...

这种方法确实完美地呈现了Swagger，但有两个主要问题与之相关：

• 当我的所有FastAPI端点返回响应时，它们都会中断并给我一个错误。例如：我的endpoint1仍然返回原始响应，但endpoint1希望它发送符合FinalResponseForEndPoint1模型的响应
• 对我的所有端点的所有模型使用这种方法似乎不是正确的方法

以下是我的自定义中间件的最小可复制示例：

from starlette.types import ASGIApp, Receive, Scope, Send, Message
from starlette.requests import Request
import json
from starlette.datastructures import MutableHeaders
from fastapi import FastAPI

class MetaDataAdderMiddleware:
application_generic_urls = ['/openapi.json', '/docs', '/docs/oauth2-redirect', '/redoc']

def __init__(
self, 
app: ASGIApp
) -> None:
self.app = app
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
if scope["type"] == "http" and not any([scope["path"].startswith(endpoint) for endpoint in MetaDataAdderMiddleware.application_generic_urls]):
responder = MetaDataAdderMiddlewareResponder(self.app, self.standard_meta_data, self.additional_custom_information)
await responder(scope, receive, send)
return
await self.app(scope, receive, send)

class MetaDataAdderMiddlewareResponder:
def __init__(
self,
app: ASGIApp,
) -> None:
"""
"""
self.app = app
self.initial_message: Message = {}
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
self.send = send
await self.app(scope, receive, self.send_with_meta_response)
async def send_with_meta_response(self, message: Message):
message_type = message["type"]
if message_type == "http.response.start":
# Don't send the initial message until we've determined how to
# modify the outgoing headers correctly.
self.initial_message = message
elif message_type == "http.response.body":
response_body = json.loads(message["body"].decode())
data = {}
data["data"] = response_body
data['metadata'] = {
'field_1': 'value_1',
'field_2': 'value_2'
}
data_to_be_sent_to_user = json.dumps(data, default=str).encode("utf-8")
headers = MutableHeaders(raw=self.initial_message["headers"])
headers["Content-Length"] = str(len(data_to_be_sent_to_user))
message["body"] = data_to_be_sent_to_user
await self.send(self.initial_message)
await self.send(message)

app = FastAPI(
title="MY DUMMY APP",
)
app.add_middleware(MetaDataAdderMiddleware)
@app.get("/")
async def root():
return {"message": "Hello World"}

Answer 1

如果向附加字段添加默认值，则可以让中间件更新这些字段，而不是创建它们。

所以：

from ast import Str
from starlette.types import ASGIApp, Receive, Scope, Send, Message
from starlette.requests import Request
import json
from starlette.datastructures import MutableHeaders
from fastapi import FastAPI
from pydantic import BaseModel, Field
# This model is common to all endpoints!
# Since we are going to add this for all responses

class MetaDataModel(BaseModel):
meta_data_1: str
meta_data_2: str
meta_data_3: str

class ResponseForEndPoint1(BaseModel):
data: str
meta_data: MetaDataModel | None = Field(None, nullable=True)

class MetaDataAdderMiddleware:
application_generic_urls = ['/openapi.json',
'/docs', '/docs/oauth2-redirect', '/redoc']
def __init__(
self,
app: ASGIApp
) -> None:
self.app = app
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
if scope["type"] == "http" and not any([scope["path"].startswith(endpoint) for endpoint in MetaDataAdderMiddleware.application_generic_urls]):
responder = MetaDataAdderMiddlewareResponder(
self.app)
await responder(scope, receive, send)
return
await self.app(scope, receive, send)

class MetaDataAdderMiddlewareResponder:
def __init__(
self,
app: ASGIApp,
) -> None:
"""
"""
self.app = app
self.initial_message: Message = {}
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
self.send = send
await self.app(scope, receive, self.send_with_meta_response)
async def send_with_meta_response(self, message: Message):
message_type = message["type"]
if message_type == "http.response.start":
# Don't send the initial message until we've determined how to
# modify the outgoing headers correctly.
self.initial_message = message
elif message_type == "http.response.body":
response_body = json.loads(message["body"].decode())
response_body['meta_data'] = {
'field_1': 'value_1',
'field_2': 'value_2'
}
data_to_be_sent_to_user = json.dumps(
response_body, default=str).encode("utf-8")
headers = MutableHeaders(raw=self.initial_message["headers"])
headers["Content-Length"] = str(len(data_to_be_sent_to_user))
message["body"] = data_to_be_sent_to_user
await self.send(self.initial_message)
await self.send(message)

app = FastAPI(
title="MY DUMMY APP",
)
app.add_middleware(MetaDataAdderMiddleware)

@app.get("/", response_model=ResponseForEndPoint1)
async def root():
return ResponseForEndPoint1(data='hello world')

我不认为这是一个好的解决方案，但它不会出现错误，而且它确实显示了正确的输出。

总的来说，我正在努力寻找一种好的方法来记录中间件可以在openAI/swagger中引入的更改/额外响应。如果你还发现什么，我很想听听！