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

Question

我正在构建一个FastAPI应用程序，它有很多Pydantic模型。尽管应用程序运行良好，但正如所料，OpenAPI (Swagger UI)文档并没有在< code>Schemas部分显示所有这些模型的模式。

下面是pydantic＜code＞模式的内容。py＜/code＞

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路径中用于验证请求体。< code>FaultQueryParams是一个自定义模型，我用它来验证请求查询参数，用法如下:

query_args:FaultQueryParams=Depends（）

其余模型与Body字段一起使用。我无法理解为什么只有一些模型在Schemas部分中不可见，而其他模型则不可见。

关于< code>FaultQueryParams,我注意到的另一件事是，描述和示例不会显示在路径endpoint上，即使它们是在模型中定义的。

编辑1:

我深入研究并意识到，在swagger UI中不可见的所有模型都不是直接用于路径操作的模型，也就是说，这些模型不是用作response_model或Body类型，而是间接使用的助手模型。因此，FastAPI似乎没有为这些模型生成模式。

上述语句的一个例外是query_args:FaultQueryParams=Depends（），它直接用于路径操作，以将endpoint的查询params映射到自定义模型。这是一个问题，因为swagger没有从该模型的字段中识别元参数，如＜code＞title＜/code＞、＜code＞description＜/code＞、＜code＞example＜/ccode＞

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

Answer 1

感谢@Chris提供的指针，这些指针最终导致我使用数据类来批量定义查询参数，而且它工作得很好。

@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",
    )

Answer 2

FastAPI将为用作请求主体或响应模型的模型生成模式。当声明query_args:FaultQueryParams=Depends（）（使用Depends）时，endpoint不需要请求体，而是需要查询parameters；因此，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

您可以让 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

...

要为查询参数声明元数据，如< code>description 、< code>example等，您应该用< code>Query而不是< code>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

上面的代码可以用< code>@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