通过 FastAPI 中的 pydantic 模型设置自定义错误响应的媒体类型

在我的 FastAPI 应用程序中，我想将错误作为 RFC Problem JSON 返回：

from pydantic import BaseModel

class RFCProblemJSON(BaseModel):
    type: str
    title: str
    detail: str | None
    status: int | None

我可以在 OpenAPI 文档中设置响应模型responsesFastAPI 类的参数：

from fastapi import FastAPI, status

api = FastAPI(
    responses={
        status.HTTP_401_UNAUTHORIZED: {'model': RFCProblemJSON},
        status.HTTP_422_UNPROCESSABLE_ENTITY: {'model': RFCProblemJSON},
        status.HTTP_500_INTERNAL_SERVER_ERROR: {'model': RFCProblemJSON}
    }
)

但是，我想将媒体类型设置为“application/problem+json”。我尝试了两种方法，首先只是在基本模型中添加“媒体类型”字段：

class RFCProblemJSON(BaseModel):
    media_type = "application/problem+json"
    type: str
    title: str
    detail: str | None
    status: int | None

并且还继承自fastapi.responses.Response:

class RFCProblemJSON(Response):
    media_type = "application/problem+json"
    type: str
    title: str
    detail: str | None
    status: int | None

但是，这些都不会修改 openapi.json 文件/swagger UI 中的 media_type 。

When you add the media_type field to the basemodel, the media type in the SwaggerUI is not modified::

当你让模型继承 Response 时，你只会得到一个错误（这是一个不太可能工作的机会，但无论如何都尝试过）。

    raise fastapi.exceptions.FastAPIError(
fastapi.exceptions.FastAPIError: Invalid args for response field! Hint: check that <class 'RoutingServer.RestAPI.schema.errors.RFCProblemJSON'> is a valid Pydantic field type. If you are using a return type annotation that is not a valid Pydantic field (e.g. Union[Response, dict, None]) you can disable generating the response model from the type annotation with the path operation decorator parameter response_model=None. Read more: https://fastapi.tiangolo.com/tutorial/response-model/

如果您手动填写 OpenAPI 定义，则可以让 swagger UI 显示正确的媒体类型：

api = FastAPI(
    debug=debug,
    version=API_VERSION,
    title="RoutingServer API",
    openapi_tags=tags_metadata,
    swagger_ui_init_oauth={"clientID": oauth2_scheme.client_id},
    responses={
        status.HTTP_401_UNAUTHORIZED: {
            "content": {"application/problem+json": {
            "example": {
                "type": "string",
                "title": "string",
                "detail": "string"
            }}},
            "description": "Return the JSON item or an image.",
        },
    }
)

但是，我想尝试使用 BaseModel 来实现此功能，以便我可以从 RFCProblemJSON 继承并为某些特定错误提供一些可选的附加功能。

重现我的问题的最小示例是：

from pydantic import BaseModel
from fastapi import FastAPI, status, Response, Request
from fastapi.exceptions import RequestValidationError
from pydantic import error_wrappers
import json
import uvicorn
from typing import List, Tuple, Union, Dict, Any
from typing_extensions import TypedDict

Loc = Tuple[Union[int, str], ...]


class _ErrorDictRequired(TypedDict):
    loc: Loc
    msg: str
    type: str


class ErrorDict(_ErrorDictRequired, total=False):
    ctx: Dict[str, Any]


class RFCProblemJSON(BaseModel):
    type: str
    title: str
    detail: str | None
    status: int | None


class RFCUnprocessableEntity(RFCProblemJSON):
    instance: str
    issues: List[ErrorDict]


class RFCProblemResponse(Response):
    media_type = "application/problem+json"

    def render(self, content: RFCProblemJSON) -> bytes:
        return json.dumps(
            content.dict(),
            ensure_ascii=False,
            allow_nan=False,
            indent=4,
            separators=(", ", ": "),
        ).encode("utf-8")


api = FastAPI(
    responses={
        status.HTTP_422_UNPROCESSABLE_ENTITY: {'model': RFCUnprocessableEntity},
    }
)


@api.get("/{x}")
def hello(x: int) -> int:
    return x


@api.exception_handler(RequestValidationError)
def format_validation_error_as_problem_json(request: Request, exc: error_wrappers.ValidationError):
    status_code = status.HTTP_422_UNPROCESSABLE_ENTITY
    content = RFCUnprocessableEntity(
        type="/errors/unprocessable_entity",
        title="Unprocessable Entity",
        status=status_code,
        detail="The request has validation errors.",
        instance=request.url.path,
        issues=exc.errors()
    )
    return RFCProblemResponse(content, status_code=status_code)


uvicorn.run(api)

当你去http://localhost:8000/hello，它将返回为application/problem+json在标题中，但是如果您转到 swagger ui 文档，ui 显示的响应将是application/json。我不知道如何保持代码的风格，但更新 openapi 定义以表明它将以一种很好的方式返回“application/problem+json”。

这可以吗？

正如 FastAPI 的文档中所述OpenAPI 中的其他响应:

您可以将参数传递给路径操作装饰器responses.

它接收到一个dict，键是每个响应的状态代码， 喜欢200，并且值是其他dicts 的信息为 他们每个人。

这些回应中的每一个dicts可以有钥匙model，包含一个 金字塔模型，就像 response_model.

FastAPI 将采用该模型，生成其 JSON 模式并包含它 在 OpenAPI 中的正确位置。

另外，如中所述模型的附加响应（见下Info):

The model key is notOpenAPI 的一部分。

FastAPI 将从那里获取 Pydantic 模型，生成JSON Schema, and 把它放在正确的地方.

正确的地方是：

• 在钥匙里content，它具有另一个 JSON 对象的值 (dict）包含：

  • 带有媒体类型的密钥，例如application/json，它包含另一个 JSON 对象作为值，该对象包含：

    • A key schema，其值为模型中的 JSON Schema，这是正确的地方.

      • FastAPI 在 OpenAPI 中的另一个位置添加了对全局 JSON 模式的引用，而不是直接包含它。这 这样，其他应用程序和客户端就可以使用这些 JSON 模式 直接，提供更好的代码生成工具等。

因此，目前似乎没有办法实现您所要求的，即添加一个media_type场到BaseModel，为了设置错误响应的媒体类型（例如，422 UNPROCESSABLE ENTITY) to application/problem+json——自从model key is only用于生成schema。已经有广泛的github 上的讨论在类似的问题上，人们提供了一些解决方案，主要集中在改变422错误响应模式，类似于您的问题中描述的模式，但以一种更优雅的方式（请参阅这条评论， 例如）。下面的示例演示了一种类似的方法，可以轻松地适应您的需求。

工作示例

from fastapi import FastAPI, Response, Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.openapi.constants import REF_PREFIX
from fastapi.responses import JSONResponse
from pydantic import BaseModel
import json


class Item(BaseModel):
    id: str
    value: str


class SubMessage(BaseModel):
    msg: str


class Message(BaseModel):
    msg: str
    sub: SubMessage


class CustomResponse(Response):
    media_type = 'application/problem+json'

    def render(self, content: Message) -> bytes:
        return json.dumps(
            content.dict(),
            ensure_ascii=False,
            allow_nan=False,
            indent=4,
            separators=(', ', ': '),
        ).encode('utf-8')


def get_422_schema():
    return {
        'model': Message,
        'content': {
            'application/problem+json': {
                'schema': {'$ref': REF_PREFIX + Message.__name__}
            }
        },
    }


app = FastAPI(responses={status.HTTP_422_UNPROCESSABLE_ENTITY: get_422_schema()})


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    msg = Message(msg='main message', sub=SubMessage(msg='sub message'))
    return CustomResponse(content=msg, status_code=status.HTTP_422_UNPROCESSABLE_ENTITY)


@app.post('/items')
async def submit(item: Item):
    return item