OpenAPI 缺少 FastAPI 应用程序中某些 Pydantic 模型的架构

Question

我正在构建一个 FastAPI 应用程序，其中有很多 Pydantic 模型。尽管应用程序运行良好，但正如预期的那样，OpenAPI (Swagger UI) 文档并未在该Schemas部分下显示所有这些模型的架构。

这是pydantic的内容schemas.py

import socket
from datetime import datetime
from enum import Enum
from typing import Any, Dict, List, Optional, Set, Union

from pydantic import BaseModel, Field, validator
from typing_extensions import Literal

ResponseData = Union[List[Any], Dict[str, Any], BaseModel]


# Not visible in Swagger UI
class PageIn(BaseModel):
    page_size: int = Field(default=100, gt=0)
    num_pages: int = Field(default=1, gt=0, exclude=True)
    start_page: int = Field(default=1, gt=0, exclude=True)

# visible under schemas on Swagger UI
class PageOut(PageIn):
    total_records: int = 0
    total_pages: int = 0
    current_page: int = 1

    class Config:  # pragma: no cover
        @staticmethod
        def schema_extra(schema, model) -> None:
            schema.get("properties").pop("num_pages")
            schema.get("properties").pop("start_page")


# Not visible in Swagger UI
class BaseResponse(BaseModel):
    host_: str = Field(default_factory=socket.gethostname)
    message: Optional[str]


# Not visible in Swagger UI
class APIResponse(BaseResponse):
    count: int = 0
    location: Optional[str]
    page: Optional[PageOut]
    data: ResponseData


# Not visible in Swagger UI
class ErrorResponse(BaseResponse):
    error: str


# visible under schemas on Swagger UI
class BaseFaultMap(BaseModel):
    detection_system: Optional[str] = Field("", example="obhc")
    fault_type: Optional[str] = Field("", example="disk")
    team: Optional[str] = Field("", example="dctechs")
    description: Optional[str] = Field(
        "",
        example="Hardware raid controller disk failure found. "
        "Operation can continue normally,"
        "but risk of data loss exist",
    )



# Not visible in Swagger UI
class FaultQueryParams(BaseModel):
    f_id: Optional[int] = Field(None, description="id for the host", example=12345, title="Fault ID")
    hostname: Optional[str]
    status: Literal["open", "closed", "all"] = Field("open")
    created_by: Optional[str]
    environment: Optional[str]
    team: Optional[str]
    fault_type: Optional[str]
    detection_system: Optional[str]
    inops_filters: Optional[str] = Field(None)
    date_filter: Optional[str] = Field("",)
    sort_by: Optional[str] = Field("created",)
    sort_order: Literal["asc", "desc"] = Field("desc")

所有这些模型实际上都在 FastAPI 路径中用于验证请求正文。这FaultQueryParams是一个自定义模型，我用它来验证请求查询参数，其使用方式如下：

query_args: FaultQueryParams = Depends()

其余模型均与现场结合使用Body。我无法弄清楚为什么只有某些模型在该Schemas部分中不可见，而其他模型则可见。

我注意到的另一件事FaultQueryParams是，描述、示例不会显示在路径端点上，即使它们是在模型中定义的。

编辑1：

我进行了更多研究并意识到，所有在 swagger UI 中不可见的模型都没有直接在路径操作中使用，即这些模型没有被用作或类型，response_model而是Body正在被使用的辅助模型。间接使用。因此，FastAPI 似乎没有为这些模型生成架构。

上述语句的一个例外是query_args: FaultQueryParams = Depends()直接在路径操作中使用它来将Query端点的参数映射到自定义模型。这是一个问题，因为 swagger 没有从该模型的字段中识别元参数，例如 、 、 ，title并且description没有example显示在 UI 上，这对于该端点的用户来说很重要。

有没有办法欺骗 FastAPI 为自定义模型生成模式，FaultQueryParams就像它为 等生成Body一样Query？

Answer 1

FastAPI 将为用作请求主体或响应模型的模型生成架构。声明时query_args: FaultQueryParams = Depends()（使用Depends），您的端点不会期望一个request body，而是期望query参数；因此，FaultQueryParams不会包含在OpenAPI文档的架构中。

要添加其他架构，您可以扩展/修改 OpenAPI 架构。下面给出了示例（确保在定义所有路由之后添加用于修改架构的代码，即在代码末尾）。

class FaultQueryParams(BaseModel):
    f_id: Optional[int] = Field(None, description="id for the host", example=12345, title="Fault ID")
    hostname: Optional[str]
    status: Literal["open", "closed", "all"] = Field("open")
    ...
    
@app.post('/predict')
def predict(query_args: FaultQueryParams = Depends()):
    return query_args

def get_extra_schemas():
    return {
              "FaultQueryParams": {
                "title": "FaultQueryParams",
                "type": "object",
                "properties": {
                  "f_id": {
                    "title": "Fault ID",
                    "type": "integer",
                    "description": "id for the host",
                    "example": 12345
                  },
                  "hostname": {
                    "title": "Hostname",
                    "type": "string"
                  },
                  "status": {
                    "title": "Status",
                    "enum": [
                      "open",
                      "closed",
                      "all"
                    ],
                    "type": "string",
                    "default": "open"
                  },
                   ...
                }
              }
            }

from fastapi.openapi.utils import get_openapi

def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema
    openapi_schema = get_openapi(
        title="FastAPI",
        version="1.0.0",
        description="This is a custom OpenAPI schema",
        routes=app.routes,
    )
    new_schemas = openapi_schema["components"]["schemas"]
    new_schemas.update(get_extra_schemas())
    openapi_schema["components"]["schemas"] = new_schemas
    
    app.openapi_schema = openapi_schema
    return app.openapi_schema


app.openapi = custom_openapi

一些有用的注释

注1

您可以让 FastAPI 使用该模型向您的代码添加一个端点（您随后将在获取架构后将其删除），从而让 FastAPI 为您完成此操作，而不是手动键入要添加到文档中的额外模型的架构作为请求正文或响应模型，例如：

@app.post('/predict') 
def predict(query_args: FaultQueryParams):
    return query_args

然后，您可以在http://127.0.0.1:8000/openapi.json获取生成的 JSON 架构，如文档中所述。从那里，您可以将模型的架构复制并粘贴到代码中并直接使用它（如get_extra_schema()上面的方法所示），或者将其保存到文件并从文件加载 JSON 数据，如下所示：

import json
...

new_schemas = openapi_schema["components"]["schemas"]

with open('extra_schemas.json') as f:    
    extra_schemas = json.load(f)
    
new_schemas.update(extra_schemas)   
openapi_schema["components"]["schemas"] = new_schemas

...

笔记2

要为您的查询参数声明元数据，例如description，example等，您应该使用而Query不是定义您的参数Field，并且由于您无法使用 Pydantic 模型执行此操作，因此您可以声明一个自定义依赖项类，如此处所述和所示以下：

from fastapi import FastAPI, Query, Depends
from typing import Optional

class FaultQueryParams:
    def __init__(
        self,
        f_id: Optional[int] = Query(None, description="id for the host", example=12345)

    ):
        self.f_id = f_id

app = FastAPI()

@app.post('/predict')
def predict(query_args: FaultQueryParams = Depends()):
    return query_args

可以使用装饰器重写上面的内容@dataclass，如下所示：

from fastapi import FastAPI, Query, Depends
from typing import Optional
from dataclasses import dataclass

@dataclass
class FaultQueryParams:
    f_id: Optional[int] = Query(None, description="id for the host", example=12345)

app = FastAPI()

@app.post('/predict')
def predict(query_args: FaultQueryParams = Depends()):
    return query_args