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

Question

我正在构建一个FastAPI应用程序，其中包含许多Pydantic模型。尽管该应用程序正常运行，但如预期的那样，OpenAPI（Swagger UI）文档在部分中并未显示所有这些模型的模式。

以下是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路径(path)以验证请求体(request body)。 FaultQueryParams 是自定义模型，我用它来验证请求查询参数(query params)，并像下面这样使用它： query_args: FaultQueryParams = Depends() 其余的模型与 Body 字段一起使用。 我无法弄清楚为什么只有其中一些模型不显示在 Schemas 部分中而其他模型则显示出来。

另外，我注意到关于 FaultQueryParams 的另一件事是，即使它们在模型中定义了，描述和示例也不会显示在路径终点(endpoint)上。

编辑1:

我进一步研究并意识到，所有在Swagger UI中不可见的模型都是那些没有直接用作路径操作的模型，即这些模型不被用作 response_model 或 Body 类型，并且是间接使用的辅助模型。因此，似乎FastAPI没有为这些模型生成架构(schema)。

以上声明的一个例外是 query_args: FaultQueryParams = Depends() ，它被直接用于路径操作以将终点(endpoint)的查询(Query)参数映射到自定义模型。 这是一个问题，因为swagger没有识别此模型的字段中的元参数（如 title, description, example ），并且未显示在UI中，这对于该终点(endpoint)的用户非常重要。

是否有一种方法可以欺骗FastAPI生成自定义模型 FaultQueryParams 的架构(schema)，就像它为 Body，Query 等生成架构(schema)？

- Rohit

Answer 1

FastAPI将为作为请求体或响应模型使用的模型生成模式。当声明query_args: FaultQueryParams = Depends()（使用Depends）时，您的终端点不会期望请求体，而是期望查询参数；因此，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为您完成。只需在代码中添加一个端点（随后会删除），使用该模型作为请求体或响应模型即可获得模式，例如：

@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

- Chris

Answer 2

感谢 @Chris 给出的指引，最终我使用了 dataclasses 来定义大量的查询参数，结果非常好。

@dataclass
class FaultQueryParams1:
    f_id: Optional[int] = Query(None, description="id for the host", example=55555)
    hostname: Optional[str] = Query(None, example="test-host1.domain.com")
    status: Literal["open", "closed", "all"] = Query(
        None, description="fetch open/closed or all records", example="all"
    )
    created_by: Optional[str] = Query(
        None,
        description="fetch records created by particular user",
        example="user-id",
    )

- Rohit