当前位置：首页 > news >正文

python使用pydantic库结合fastApi实现接口参数验证

news 来源：原创 2025/6/20 11:48:15

pydantic库介绍

pydantic 是一个基于 Python 类型提示（Type Hints）的数据验证和解析库，主要用于确保输入数据的结构和类型符合预期。它通过定义数据模型自动完成验证、类型转换和序列化，广泛应用于 API 开发、配置管理和数据处理等领域，主要有以下几个特点：

提供类型安全的数据验证
简化复杂数据结构的处理
与 FastAPI 等现代框架深度集成

安装pydantic

使用pip安装Pydantic
```
pip install pydantic
```

pydantic核心特性

基于类型提示的模型定义

使用 Python 原生类型注解（如 str、int、List）定义字段类型，支持嵌套模型和复杂数据结构

示例
```
from pydantic import BaseModel
class User(BaseModel):id: intname: str = "John Doe"  # 默认值email: str | None = None  # 可选字段
```
自动化数据验证

类型检查：自动验证输入数据是否符合字段类型（如字符串长度、数值范围）

自定义验证器：通过 @validator 装饰器实现复杂逻辑（如密码匹配、邮箱格式）
```
from pydantic import field_validator
class Product(BaseModel):price: float@field_validator("price")def validate_price(cls, v):if v <= 0:raise ValueError("价格必须为正数")return v
```
数据转换与序列化

自动类型转换：如将字符串 "123" 转为整数 123

序列化支持：通过 .dict() 或 .json() 方法将模型转为字典或 JSON 字符串
错误处理与文档生成

详细错误报告：验证失败时返回字段级错误信息（如 "age must be ≥ 18"）

OpenAPI 集成：与 FastAPI 配合自动生成 API 文档
高性能

核心验证逻辑用 Rust 编写，速度优于纯 Python 实现的库（如 Marshmallow）

pydantic结合fastApi使用

定义请求参数模型，模型是可以继承的，以站点为例

from pydantic import BaseModel, Field, field_validator
from typing import List, Optional
from datetime import dateclass StationBase(BaseModel):station_group_id: int = Field(..., gt=0, example=1, description="站点组ID")station_name: str = Field(...,min_length=1,max_length=64,example="上海虹桥站",description="站点名称（1-64字符）",)location: Optional[str] = Field(None, max_length=128, example="上海市闵行区")online_number: int = Field(..., ge=0, example=10, description="在线设备数量")total_number: int = Field(..., gt=0, example=20, description="总设备数量")power: float = Field(..., gt=0, example=100.5, description="装机功率(kW)")capacity: float = Field(..., gt=0, example=200.0, description="装机容量(kWh)")commissioning_date: date = Field(..., example="2025-01-01")user_list: List[int] = Field(default=[], example=[1, 2], description="关联用户ID列表")class StationCreate(StationBase):passclass StationEdit(StationBase):station_id: int = Field(..., gt=0, example=1, description="要修改的站点ID")class StationDelete(BaseModel):station_id: int = Field(..., gt=0, example=1, description="要删除的站点ID")class StationQuery(BaseModel):user_id: int = Field(..., gt=0, example=1, description="用户ID")class StationGet(BaseModel):station_id: int = Field(..., gt=0, example=1, description="站点ID")

fastApi接口里面加入请求参数模型

@router.post("/add_station", tags=["Station"])
async def add_station(request: Request,station_create: StationCreate,station_dao: StationDao = Depends(StationDao),
):# 其他内容if await station_dao.add_station(station_create.station_group_id,station_create.station_name,station_create.location,station_create.online_number,station_create.total_number,station_create.power,station_create.capacity,station_create.commissioning_date,station_create.user_list,):return JSONResponse({"status": 200,"msg": "添加站点成功!",})else:return JSONResponse({"status": 400,"msg": "添加站点失败!",})# 其他内容@router.post("/delete_station", tags=["Station"])
async def delete_station(request: Request,station_delete: StationDelete,station_dao: StationDao = Depends(StationDao),
):# 其他内容# 其余接口

自定义全局错误消息返回

from fastapi.exceptions import RequestValidationError@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):errors = []for error in exc.errors():field = error["loc"][-1]msg = error.get("msg", "参数错误")errors.append({"field": field, "message": msg})return JSONResponse(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,content={"status": 422, "msg": "参数校验失败", "errors": errors},)

样例

pydantic模型里面设置某个参数限制为正数

class StationBase(BaseModel):# 其他online_number: int = Field(..., ge=0, example=10, description="在线设备数量")total_number: int = Field(..., gt=0, example=20, description="总设备数量")# 其他