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

2023-12-21

我正在构建一个 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问题在于,即使在模型中定义了描述、示例,也不会针对路径端点显示它们。

Edit 1:

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

上述声明的一个例外是query_args: FaultQueryParams = Depends()它直接用于路径操作来映射Query针对自定义模型的端点的参数。这是一个问题,因为 swagger 无法识别元参数,例如title, description, example来自此模型的字段,并且未显示在 UI 上,这对于此端点的用户很重要。

有没有办法欺骗 FastAPI 为自定义模型生成架构FaultQueryParams就像它生成的一样Body, Query etc ?


FastAPI 将为模型生成模式,这些模式可以用作请求正文 https://fastapi.tiangolo.com/tutorial/body/ or 响应模型 https://fastapi.tiangolo.com/tutorial/response-model/。申报时query_args: FaultQueryParams = Depends() (using Depends https://fastapi.tiangolo.com/tutorial/dependencies/),您的端点不会期望request body, 反而query参数;因此,FaultQueryParams不会包含在架构中OpenAPI https://fastapi.tiangolo.com/tutorial/first-steps/#openapi docs.

要添加其他模式,您可以扩展/修改 OpenAPI 架构 https://fastapi.tiangolo.com/advanced/extending-openapi/#modify-the-openapi-schema。下面给出了示例(确保在定义所有路由之后添加用于修改架构的代码,即在代码末尾)。

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

一些有用的注释

Note 1

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

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

然后,您可以在以下位置获取生成的 JSON 模式:http://127.0.0.1:8000/openapi.json http://127.0.0.1:8000/openapi.json,如中所述文档 https://fastapi.tiangolo.com/tutorial/first-steps/#check-the-openapijson。从那里,您可以将模型的架构复制并粘贴到您的代码中并直接使用它(如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

...

Note 2

To 声明元数据 https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#declare-more-metadata, 例如description, example等,对于您的查询参数,您应该使用以下方式定义您的参数Query代替Field,并且由于您无法使用 Pydantic 模型做到这一点,因此您可以声明一个自定义依赖类 https://fastapi.tiangolo.com/tutorial/dependencies/classes-as-dependencies/#classes-as-dependencies_1,如上所述here https://stackoverflow.com/a/64366434并如下图所示:

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 https://docs.python.org/3/library/dataclasses.html装饰器,如下图:

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
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)

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

随机推荐