前言

前段时间接手了算法项目，项目中的接口以及函数没有定义出参入参的Schema，这对代码的阅读和接口调试有非常大的坏处，从而导致代码优化和后续重构工作举步维艰

众所周知在工程开发中，清晰定义出参入参的类型，有很多好处：

• 提升代码可读性与可维护性，降低阅读成本，减少团队开发摩擦
• 有利于文档自动化生成
• 便于重构与迭代，提高重构时的信心
• 结合框架实现自动校验参数正确性

不同于Java这些编译性语言，可以在编译时即可以检测出类型不匹配的问题，具有良好的工程性

Python作为一门解释型语言采用的是“鸭子类型”实现多态，并不会强制要求类型匹配，如此灵活性适合快速实现复杂的算法流程

但这种灵活性也使得Python在工程化上存在一些缺陷，这也是牺牲可维护性和可读性的代价

本文主要分享笔者在Python算法服务开发时，如何使用Pydantic的一些经验

Python的类型暗示

为了提高代码的可阅读与可维护性，Python提供了TypeHint​(类型暗示)机制，主要利用typing​模块

详细使用方法可以参考官网：docs.python.org/3/library/typing.html

下面是一些常用的例子

from typing import Literal, Union, List, Dict, Any, Optional


# 对成员字段进行类型注释
class Product:
    def __init__(self):
        self.name: str
        self.pid: str
        self.level: Literal[1, 2, 3] # only allow these values
        self.params: Union[int|List[int]] # int or list of ints
        self.extra_info: Optional[Dict[str, Any]] # optional dict with any keys and values, or None

# 对函数参数和返回值进行类型注释
def get_product(product_id: int, product_name: str, level: Literal[1, 2, 3] ) -> Optional[Product]:
    pass

大部分IDE(集成开发工具)都会从分利用该机制告知开发者，提前避免潜在的问题，如下图

​typing​还支持泛型

from typing import TypeVar, Generic, List

class Data:
    pass

# 定义类型变量
T = TypeVar("T", bound=Data)  # 必须是 Data 的子类或 Data 本身
U = TypeVar("U")  # 可以是任意类型


# 泛型栈类
class Stack(Generic[T]):
    def __init__(self) -> None:
        self._items: List[T] = []

    def __getitem__(self, item) -> T:
        """向栈中添加元素"""
        return self._items[item]

int_stack: Stack[Data] = Stack()

如果希望输入的类本身，可以如下进行类型暗示

from typing import TypeVar, Type

class Data:
    pass

T = TypeVar('T', bound=Data)

def fun(data_cls: Type[T]) -> T:
    return data_cls()

基于Pydantic实现Schema

虽然Python提供了基于typing包的类型暗示机制，但仅仅只起到了提示作用，编译器并不会对此进行检查和报错，而Pydantic则可以基于类型暗示进行检验，当类型不匹配时会报错

定义与使用

Pydantic的BaseModel​类可以方便实现Schema，以下是一个示例

from pydantic import BaseModel, Field
from typing import  Optional

class DocxFile(BaseModel):
    file_name: Optional[str] = Field(None, description="The name of the DOCX file.") # Optional[str] 表示这个字段是可选的，并且类型为字符串
    file_link: str = Field(..., description="The link to the DOCX file.") # 这个字段是必需的。
    # Field description 不是必须的，但是fastapi可以利用它来生成文档

# 实例化时使用关键字参数
docx_file = DocxFile(file_name="example.docx", file_link="example.com/example.docx")
print(docx_file.file_name, docx_file.file_link) # 直接访问成员变量获取信息

嵌套的Schema也可以实现

from pydantic import BaseModel, Field
from typing import  Optional, List

class DocxFile(BaseModel):
    file_name: Optional[str] = Field(None, description="The name of the DOCX file.")
    file_link: str = Field(..., description="The link to the DOCX file.")

class PdfFile(BaseModel):
    file_name: Optional[str] = Field(None, description="The name of the PDF file.")
    file_link: str = Field(..., description="The link to the PDF file.")
    is_scanned: bool = Field(False, description="Indicates if the PDF is scanned.")

class Author(BaseModel):
    author_name: str
    author_id: int

class FileInfo(BaseModel):
    file_id: int # 必填字段
    file_name: str # 必填字段
    docx_file: Optional[DocxFile] = None
    pdf_file: Optional[PdfFile] = None
    author: List[Author]

file_info = FileInfo(
    file_id=123,
    file_name="example",
    author=[
        Author(author_name="John Doe", author_id=1),
    ],
    docx_file=DocxFile(
        file_name="example.docx",
        file_link="example.com/example.docx"
    ),
)

print(file_info.docx_file.file_name)

如果嵌套类较多，建议把外层类放到代码最顶层，符合自上而下的阅读习惯

class FileInfo(BaseModel):
    file_id: int # 必填字段
    file_name: str # 必填字段
    docx_file: Optional['DocxFile'] = None
    pdf_file: Optional['PdfFile'] = None
    author: List['Author']

class DocxFile(BaseModel):
    pass

class PdfFile(BaseModel):
    pass

class Author(BaseModel):
    pass

校验

重复Pydantic的校验功能，可以将校验代码与主逻辑代码解耦，减少算法代码中的重复检验代码

实例化BaseModel时候，会自动根据自动的类型暗示进行校验，typing关键字Literal、Uion、Dict、Any都支持，如果校验失败会报错

from pydantic import BaseModel, ValidationError, Field
from typing import Optional

class User(BaseModel):
    username: str = Field(..., min_length=2, max_length=20) # 用户名，必须是字符串且长度在2到20之间
    age: int = Field(..., gt=0, lt=150) # 年龄，必须是整数且大于0小于150
    email: Optional[str] = Field(
        None, pattern=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
    ) # 邮箱地址，可选字段，如果提供则必须符合邮箱格式


def parse_validation_error(e: ValidationError) -> str:
	"""软色报错"""
    result_msg = f"请求参数错误，发现{len(e.errors())}个字段错误\n"
    for error in e.errors():
        loc = error.get("loc", [])
        # 将 loc 中的元素转换为字符串，并用 "." 拼接，数字用 [index] 表示
        formatted_loc = ".".join(
            f"[{str(part)}]" if isinstance(part, int) else str(part) for part in loc
        ).replace(".[", "[")
        msg = error.get("msg", "")
        input_value = error.get("input", "")
        expected_values = error.get("ctx", {}).get("expected", "")

        result_msg += f"字段的{formatted_loc}的输入值{input_value}错误"
        if expected_values and "Input should be" in msg:
            result_msg += f", 期望值:[{expected_values}]\n"
        elif not expected_values:
            result_msg += f", 报错信息:{msg}\n"
        else:
            result_msg += f", 期望值:[{expected_values}], 报错信息:{msg}\n"
    return result_msg

try:
    user3 = User(username="王", age=30, email="wangwu@example.com")
except ValidationError as e:
    print(parse_validation_error(e)) 
"""
输出报错：
请求参数错误，发现1个字段错误
字段的username的输入值王错误, 报错信息:String should have at least 2 characters
"""

可以结合枚举，支持json转换为枚举，这一点在定义接口Schema时很有用

from enum import Enum
from pydantic import BaseModel
import json

class UserRole(str, Enum):
    ADMIN = "admin"
    USER = "user"
    GUEST = "guest"

class User(BaseModel):
    name: str
    role: UserRole

User(**json.loads('{"name":"Alice","role":"admin"}'))

也可以实现自定义校验，model_validator有多种模式，详细可以看@model_validator​的源码

from pydantic import BaseModel, model_validator

class Square(BaseModel):
    width: float
    height: float

    @model_validator(mode="after")
    def verify_square(self) -> 'Square':
        if self.width != self.height:
            raise ValueError("width and height do not match")
        return self

s = Square(width=1, height=2)
print(repr(s))

转换

​BaseModel​类可以与字典之间的转换，也可以转换为json字符串

from pydantic import BaseModel

class Schema(BaseModel):
    name: str

schema = Schema(name='John')
# basemodel=>字典
print(schema.model_dump())

# 字典=>basemodel
print(Schema(**{'name': 'John'}))

# basemodel=>json
print(schema.model_dump_json())

Pydantic在Fastapi中集成使用

简单参数校验

下面是基于Pydantic定义入参的接口，基于此代码展示如何显示入参的哪个字段存在检验错误

from typing import List

from pydantic import BaseModel, Field
from typing import Optional
import uvicorn
from fastapi import FastAPI


app = FastAPI()

class TestSchema(BaseModel):
    class TestCase(BaseModel):
        case_id: str = Field(..., description="case的id")
        params: List[str] = Field(list, description="测试参数")

    id: str
    case: List[TestCase] = Field(list, description="case列表")
    date: Optional[str] = Field(None, description="测试日期")


@app.post("/test", summary="测试接口")
async def test(
    test_name: str, model: TestSchema
) -> bool:
    return True

uvicorn.run(app, port=8100)

现在以下面的请求体发送请求应该会报错，因为id和case_id应该是整形

{
  "id": 1, # 错误类型
  "case": [
    {
      "case_id": 1, # 错误类型
      "params": [
        "string"
      ]
    }
  ],
  "date": "string"
}

接口是返回了422，响应体如下，这种响应体阅读体验较差，通常会被调用方诟病

{
  "detail": [
    {
      "type": "string_type",
      "loc": [
        "body",
        "id"
      ],
      "msg": "Input should be a valid string",
      "input": 1
    },
    {
      "type": "string_type",
      "loc": [
        "body",
        "case",
        0,
        "case_id"
      ],
      "msg": "Input should be a valid string",
      "input": 1
    }
  ]
}

为了美化报错信息，本文提供转换代码parse_request_validation_error​，实现拦截器validate_interceptor​

def validate_interceptor(request: Request, exc: RequestValidationError) -> JSONResponse:
    """
    拦截接口关于schema检验的报错，并返回约定的body
    """

    reason: str = parse_request_validation_error(exc)
    return JSONResponse(
        status_code=200,  # 或其他合适的状态码
        content=ResponseSchema(success=False, msg=reason).model_dump(),  # 将 Pydantic 模型转换为字典
    )


def parse_request_validation_error(exc: RequestValidationError):
    """
    报错润色
    RequestValidationError([{'type': 'string_type', 'loc': ('body', 'messages', 0, 'role'), 'msg': 'Input should be a valid string...[]},
    {'type': 'string_type', 'loc': ('body', 'messages', 0, 'content'), 'msg': 'Input should be a valid string', 'input': []}])
    转换结果如下：
    '请求参数错误，发现2个字段问题：messages[0].role: Input should be a valid string；messages[0].content: Input should be a valid string'
    """
    error_fields = []
    for error in exc.errors():
        loc = error["loc"]
        # 忽略第一个元素（通常是 body）
        field_path = ".".join(
            f"[{i}]" if isinstance(i, int) else i for i in loc[1:]
        ).replace(".[", "[")
        msg = error["msg"]
        error_fields.append(f"{field_path}: {msg}")
    return f"请求参数错误，发现{len(error_fields)}个字段问题：" + "；".join(
        error_fields
    )

注册validate_interceptor​

@app.exception_handler(RequestValidationError)
async def generic_exception_handler(request: Request, exc: RequestValidationError):
    return validate_interceptor(request=request, exc=exc)

重新请求，报错如下

{
  "success": false,
  "msg": "请求参数错误，发现2个字段问题：id: Input should be a valid string；case[0].case_id: Input should be a valid string"
}

自动化文档

Fastapi基于OpenAPI规范生成两种风格文档：

• 开发调试建议用 /docs​，可以直接调试接口
• 文档交付用 /redoc​，可读性更高

早在2020年时，笔者基于Fastapi开发一些算法项目，当时Pydantic处于0.x版本，Fastapi和Pydantic的关系远不如如今紧密，接收复杂的请求体时只能使用字典类型，其校验代码需要额外编写

此外Fastapi生成的Swagger文档中，无法显示多层嵌套的请求体参数形式，只能在router的函数注释中编写markdown

如今Pydantic已经更新到2.x版本，Fastapi似乎已经深度集成Pydantic，基于Pydantic的校验和文档自动化的能力已经非常完善

下面是基于Pydantic定义入参的接口

/docs文档

/redoc文档

最后

虽然说并不强制使用类型暗示，但是涉及到团队合作的项目，善于利用类型暗示可以减少开发摩擦，提高项目的健壮性

基于类型暗示实现的IDE提示、Pydantic校验、Fastapi自动化文档等功能，都实实在在提高了Python的工程的开发体验

最后，欢迎各位大佬留言，对本文提出宝贵建议，共同进步