前言
前段時間接(jie)手了算(suan)法項(xiang)目,項(xiang)目中的(de)接(jie)口以(yi)及函數沒有定義出參入參的(de)Schema,這對代碼(ma)的(de)閱讀和接(jie)口調試有非(fei)常大(da)的(de)壞處,從(cong)而(er)導致代碼(ma)優化和后續重(zhong)構工作舉(ju)步維艱
眾所周知(zhi)在工程開發中(zhong),清晰定義(yi)出參(can)(can)入參(can)(can)的類型,有(you)很多好處(chu):
- 提升代碼可讀性與可維護性,降低閱讀成本,減少團隊開發摩擦
- 有利于文檔自動化生成
- 便于重構與迭代,提高重構時的信心
- 結合框架實現自動校驗參數正確性
不同于Java這些編譯性語言(yan),可(ke)以(yi)在編譯時即可(ke)以(yi)檢測出(chu)類型(xing)不匹(pi)配的問題(ti),具有良好的工程性
Python作為一(yi)門(men)解釋(shi)型語(yu)言采用的(de)是“鴨子類型”實現多態,并不(bu)會強制要求類型匹(pi)配(pei),如此靈活性適合快速實現復雜的(de)算法(fa)流程
但(dan)這(zhe)種靈(ling)活性(xing)也(ye)使得(de)Python在工程化(hua)上(shang)存在一些缺陷,這(zhe)也(ye)是犧牲可(ke)維護性(xing)和可(ke)讀性(xing)的代價
本文主要分(fen)享(xiang)筆者在(zai)Python算(suan)法(fa)服務開發時,如何使用Pydantic的一些經驗(yan)
Python的類型暗示
為了提高代碼的可閱讀與可維護(hu)性,Python提供了TypeHint?(類型暗示(shi))機制,主要利用typing?模塊
詳細(xi)使(shi)用(yong)方(fang)法可以參(can)考官網:docs.python.org/3/library/typing.html
下面是一些常用的(de)例(li)子
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
大(da)部分IDE(集(ji)成開發(fa)工(gong)具)都會從分利(li)用該機制告知(zhi)開發(fa)者,提前避免(mian)潛在的(de)問題,如下(xia)圖

?typing?還支持(chi)泛(fan)型(xing)
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()
如(ru)果希望輸入的類本身,可以如(ru)下進行類型暗示
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
雖然(ran)Python提供了基于(yu)typing包的(de)類(lei)型(xing)暗示(shi)機(ji)制(zhi),但僅僅只起到了提示(shi)作用,編(bian)譯器并不(bu)會對此進行檢查(cha)和(he)報(bao)錯,而Pydantic則可以基于(yu)類(lei)型(xing)暗示(shi)進行檢驗,當類(lei)型(xing)不(bu)匹配時會報(bao)錯
定義與使用
Pydantic的BaseModel?類(lei)可以(yi)方(fang)便實現Schema,以(yi)下是一個示例(li)
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) # 直接訪問成員變量獲取信息
嵌套(tao)的Schema也可以(yi)實現
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)
如果嵌套類(lei)較多,建議把外層(ceng)類(lei)放到代(dai)碼最頂層(ceng),符合自上(shang)而(er)下的閱讀習慣
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的校驗(yan)功能(neng),可以將(jiang)校驗(yan)代(dai)(dai)碼與(yu)主(zhu)邏(luo)輯代(dai)(dai)碼解耦,減少算法代(dai)(dai)碼中的重復檢驗(yan)代(dai)(dai)碼
實例化BaseModel時候,會(hui)(hui)自動(dong)根據自動(dong)的(de)類(lei)型暗示進行(xing)校驗(yan),typing關鍵字Literal、Uion、Dict、Any都支持,如果(guo)校驗(yan)失敗會(hui)(hui)報錯
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
"""
可以結(jie)合枚舉,支(zhi)持json轉換為枚舉,這一點在(zai)定義接口Schema時很有用(yong)
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"}'))
也可(ke)以(yi)實(shi)現自(zi)定義(yi)校驗,model_validator有(you)多種模式,詳細(xi)可(ke)以(yi)看@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?類(lei)可以與字典之間的轉換,也可以轉換為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定義入參的接口,基于此(ci)代碼展示(shi)如(ru)何顯示(shi)入參的哪(na)個字段(duan)存在檢驗錯誤
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)
現(xian)在以(yi)下面的請求體發(fa)送請求應該會報錯,因(yin)為id和(he)case_id應該是(shi)整(zheng)形(xing)
{
  "id": 1, # 錯誤類型
  "case": [
    {
      "case_id": 1, # 錯誤類型
      "params": [
        "string"
      ]
    }
  ],
  "date": "string"
}
接口是(shi)返(fan)回了422,響應體如下,這種響應體閱(yue)讀體驗較(jiao)差,通(tong)常會被調(diao)用(yong)方詬病
{
  "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
    }
  ]
}
為了美(mei)化(hua)報錯信息(xi),本文提供轉換代碼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
    )
注(zhu)冊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規范生成兩種風格文檔(dang):
- 開發調試建議用 /docs?,可以直接調試接口
- 文檔交付用 /redoc?,可讀性更高
早在(zai)2020年時(shi),筆(bi)者(zhe)基于Fastapi開發一(yi)些算法項目,當(dang)時(shi)Pydantic處于0.x版本,Fastapi和Pydantic的關系(xi)遠不如如今(jin)緊密,接(jie)收復雜的請求(qiu)體(ti)時(shi)只能使用字典類(lei)型,其校驗代碼(ma)需要(yao)額(e)外(wai)編寫
此外(wai)Fastapi生成(cheng)的Swagger文檔中(zhong),無法顯示(shi)多層嵌套的請求體參(can)數(shu)形式,只能在router的函數(shu)注釋中(zhong)編(bian)寫markdown

如今Pydantic已經更新(xin)到2.x版本,Fastapi似乎已經深度(du)集(ji)成(cheng)Pydantic,基于(yu)Pydantic的校驗和文(wen)檔自動化的能力已經非(fei)常完善
下面(mian)是基(ji)于Pydantic定義入參的(de)接口
/docs文檔

/redoc文檔

最后
雖然說并不強制使用類(lei)型暗(an)示,但是涉及(ji)到團隊合作的項目(mu),善(shan)于(yu)利用類(lei)型暗(an)示可以減少開發摩(mo)擦(ca),提高項目(mu)的健(jian)壯(zhuang)性
基于類型暗示實(shi)現的IDE提(ti)示、Pydantic校驗、Fastapi自(zi)動(dong)化文檔等(deng)功(gong)能,都實(shi)實(shi)在在提(ti)高了Python的工程的開(kai)發體驗
最后,歡迎各位大佬留言,對本文提(ti)出寶貴建議,共同(tong)進步