由于自定义中间件，FastAPI swagger 不呈现？

Question

所以我有一个这样的自定义中间件：

它的 objective 是为我的 FastAPI 应用程序的所有端点的每个响应添加一些 meta_data 字段。


@app.middelware("http")
async def add_metadata_to_response_payload(request: Request, call_next):

    response = await call_next(request)

    body = b""
    async for chunk in response.body_iterator:
        body+=chunk


    data = {}
    data["data"] = json.loads(body.decode())
    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"
    }

    body = json.dumps(data, indent=2, default=str).encode("utf-8")

    return Response(
        content=body,
        status_code=response.status_code,
        media_type=response.media_type
    )

但是，当我使用 uvicorn 为我的应用程序提供服务并启动 swagger URL 时，这是我看到的：


Unable to render this definition

The provided definition does not specify a valid version field.

Please indicate a valid Swagger or OpenAPI version field. Supported version fields are
Swagger: "2.0" and those that match openapi: 3.0.n (for example, openapi: 3.0.0)

经过大量调试，我发现这个错误是由于自定义中间件引起的，特别是这一行：

body = json.dumps(data, indent=2, default=str).encode("utf-8")

如果我简单地注释掉这一行，swagger 渲染对我来说就很好。但是，我需要这一行来传递来自中间件的响应中的内容参数。如何解决这个问题？

更新：

我尝试了以下方法： body = json.dumps(data, indent=2).encode("utf-8") 通过删除默认参数，swagger 成功加载。但是现在，当我点击任何 API 时，这是 swagger 告诉我的以及屏幕上的响应负载： Unrecognised response type; displaying content as text

更多更新（2022 年 4 月 6 日）：

Chris 找到了解决问题的一部分的解决方案，但仍然没有加载 swagger。代码无限期挂在中间件级别，页面还没有加载。

所以，我在所有这些地方找到了：

https://github.com/encode/starlette/issues/919
Blocked code while using middleware and dependency injections to log requests in FastAPI(Python)
https://github.com/tiangolo/fastapi/issues/394

这种添加自定义中间件的方式通过继承 Starlette 中的 BaseHTTPMiddleware 来工作，并且有其自身的问题（与等待内部中间件、streamingresponse 和正常响应以及调用方式有关）。还没看懂

Answer 1

您正在用取自中间件和响应（在本例中为 html 响应）的 json 数据替换 swagger 的主体 html。

你最终会得到类似

的东西

{
    "data": "<html>....</html>",
    "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"
    }
}

当然不行。

可能的解决方案

检查中间件中响应的内容类型。扩展响应 json，否则保持原样。

注意：仅当可以安全地假设每个 json 响应都需要添加 metadata 而 html 内容类型不需要时，才能这样做。（您可以根据需要更改支票）

另一种可能的解决方案

等待以下问题合并到当前 starlette 的实现中，然后 fastapi 开始使用此版本。

https://github.com/tiangolo/fastapi/issues/1174 https://github.com/encode/starlette/pull/1286

Answer 2

这是您可以做到的方法（灵感来自 this）。一定要勾选response的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 路径（针对您想要的 paths/routes 的 pre-defined 列表将元数据添加到他们的响应中），并相应地进行：

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"}

由于自定义中间件，FastAPI swagger 不呈现？

FastAPI swagger does not render because of custom Middleware?

python

fastapi

可能的解决方案

另一种可能的解决方案

更新 1

更新 2

工作示例