在Python开发中，数据验证是一个永恒的话题。无论是处理用户输入、解析API响应，还是配置管理，确保数据的正确性和完整性都是至关重要的。Pydantic作为Python生态中最流行的数据验证库，以其优雅的API设计、出色的性能和与类型系统的深度集成，成为了FastAPI等现代框架的核心依赖。

本文将全面介绍Pydantic的使用方法，从基础模型定义到高级验证技巧，帮助你在实际项目中充分发挥Pydantic的威力。

一、Pydantic基础

1.1 为什么选择Pydantic

在Pydantic出现之前，Python开发者通常使用以下方式验证数据：

# 传统方式：手动验证
def create_user(data):
    if not isinstance(data.get('name'), str) or len(data['name']) < 2:
        raise ValueError("姓名必须是至少2个字符的字符串")
    if not isinstance(data.get('age'), int) or data['age'] < 0:
        raise ValueError("年龄必须是正整数")
    if data.get('email') and '@' not in data['email']:
        raise ValueError("邮箱格式不正确")
    # ... 更多验证逻辑

这种方式代码冗长、难以维护，且缺乏统一的错误处理。Pydantic提供了一种声明式的数据验证方式：

from pydantic import BaseModel, EmailStr, Field, field_validator

class User(BaseModel):
    name: str = Field(min_length=2, max_length=50)
    age: int = Field(gt=0, le=150)
    email: EmailStr

    @field_validator('name')
    @classmethod
    def name_must_contain_space(cls, v):
        if ' ' not in v:
            raise ValueError('姓名必须包含空格（姓和名）')
        return v.title()

# 使用
user = User(name="张 三", age=25, email="zhangsan@example.com")
print(user.model_dump())  # {'name': '张 三', 'age': 25, 'email': 'zhangsan@example.com'}

1.2 安装与基本配置

pip install pydantic[email]  # 包含EmailStr支持

Pydantic V2在性能上相比V1有了巨大提升，核心验证逻辑使用Rust重写，性能提升了5-50倍。

二、模型定义与字段类型

2.1 基本类型

Pydantic支持Python的所有标准类型，并提供了额外的验证类型：

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

class Article(BaseModel):
    title: str
    content: str
    author: str
    tags: List[str] = []
    published: bool = False
    published_at: Optional[datetime] = None
    views: int = 0

article = Article(
    title="Pydantic入门指南",
    content="这是一篇关于Pydantic的文章...",
    author="张三",
    tags=["python", "validation", "tutorial"]
)

2.2 嵌套模型

实际应用中，数据通常是嵌套结构的。Pydantic完美支持嵌套模型：

from pydantic import BaseModel
from typing import List

class Address(BaseModel):
    province: str
    city: str
    district: str
    street: str
    zip_code: str = Field(pattern=r'^\d{6}$')

class Company(BaseModel):
    name: str
    industry: str
    address: Address
    departments: List[str]

class Employee(BaseModel):
    name: str
    position: str
    company: Company
    skills: List[str]

# 使用嵌套模型
employee = Employee(
    name="李四",
    position="高级工程师",
    company={
        "name": "科技有限公司",
        "industry": "互联网",
        "address": {
            "province": "北京市",
            "city": "北京市",
            "district": "海淀区",
            "street": "中关村大街1号",
            "zip_code": "100080"
        },
        "departments": ["技术部", "产品部"]
    },
    skills=["Python", "FastAPI", "Docker"]
)

2.3 字段验证器

from pydantic import BaseModel, field_validator, model_validator

class Product(BaseModel):
    name: str
    price: float
    discount: float = 0.0
    stock: int

    @field_validator('price', 'discount')
    @classmethod
    def check_positive(cls, v):
        if v < 0:
            raise ValueError('价格不能为负数')
        return round(v, 2)

    @field_validator('name')
    @classmethod
    def check_name_length(cls, v):
        if len(v.strip()) < 2:
            raise ValueError('产品名称至少2个字符')
        return v.strip()

    @model_validator(mode='after')
    def check_discount_not_exceed_price(self):
        if self.discount >= self.price:
            raise ValueError('折扣价不能大于或等于原价')
        return self

三、高级用法

3.1 配置管理

from pydantic_settings import BaseSettings
from pydantic import Field

class AppConfig(BaseSettings):
    app_name: str = "MyApp"
    debug: bool = False
    database_url: str = Field(alias="DATABASE_URL")
    redis_url: str = Field(default="redis://localhost:6379")
    jwt_secret: str = Field(min_length=32)
    jwt_expire_minutes: int = Field(default=30, gt=0)

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

# 从环境变量和.env文件自动加载配置
config = AppConfig()
print(config.app_name)
print(config.database_url)

3.2 自定义数据类型

from pydantic import BaseModel, GetCoreSchemaHandler
from pydantic_core import core_schema
from typing import Any

class PhoneNumber(str):
    """自定义手机号类型，自动格式化和验证"""

    @classmethod
    def __get_pydantic_core_schema__(
        cls, source_type: Any, handler: GetCoreSchemaHandler
    ):
        return core_schema.with_info_plain_validator_function(
            cls._validate
        )

    @classmethod
    def _validate(cls, value: str, _info) -> 'PhoneNumber':
        value = value.strip().replace('-', '').replace(' ', '')
        if not value.isdigit() or len(value) != 11:
            raise ValueError('请输入有效的11位手机号码')
        if not value.startswith('1'):
            raise ValueError('手机号必须以1开头')
        return cls(value)

class ContactForm(BaseModel):
    name: str
    phone: PhoneNumber
    message: str

3.3 数据转换与序列化

from pydantic import BaseModel, ConfigDict
from datetime import datetime
from enum import Enum

class Status(str, Enum):
    ACTIVE = "active"
    INACTIVE = "inactive"
    PENDING = "pending"

class UserRecord(BaseModel):
    model_config = ConfigDict(
        from_attributes=True,  # 支持从ORM对象创建
        populate_by_name=True,  # 支持别名
    )

    id: int
    username: str
    status: Status
    created_at: datetime

    def to_api_response(self) -> dict:
        """自定义API响应格式"""
        return {
            'user_id': self.id,
            'name': self.username,
            'status': self.status.value,
            'created': self.created_at.isoformat()
        }

四、与FastAPI深度集成

4.1 请求体验证

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, EmailStr, Field

app = FastAPI()

class CreateOrderRequest(BaseModel):
    customer_name: str = Field(min_length=2, max_length=100)
    customer_email: EmailStr
    items: list[OrderItem]
    shipping_address: Address
    notes: str | None = None

    @field_validator('items')
    @classmethod
    def validate_items(cls, v):
        if len(v) == 0:
            raise ValueError('订单至少包含一个商品')
        return v

@app.post("/api/orders")
async def create_order(request: CreateOrderRequest):
    """FastAPI自动验证请求体"""
    order = await order_service.create(request)
    return {"order_id": order.id, "status": "created"}

4.2 响应模型

from fastapi import FastAPI
from pydantic import BaseModel

class OrderResponse(BaseModel):
    order_id: str
    total_amount: float
    status: str
    created_at: str

@app.get("/api/orders/{order_id}", response_model=OrderResponse)
async def get_order(order_id: str):
    order = await order_service.get(order_id)
    return order  # 自动序列化为OrderResponse格式

五、性能优化技巧

1. 使用model_validate代替构造函数

# 推荐：使用model_validate
data = {"name": "test", "age": 25}
user = User.model_validate(data)

# 对于JSON字符串
import json
user = User.model_validate_json(json_str)

2. 避免不必要的验证

# 当数据来源可信时，可以跳过验证
user = User.model_construct(name="test", age=25)

3. 使用strict模式提升性能

class StrictModel(BaseModel):
    model_config = ConfigDict(strict=True)
    count: int  # 不会自动将"123"转换为123

结语

Pydantic已经从一个简单的数据验证库发展成为Python数据处理的基石。它与Python类型系统的深度集成、出色的性能表现以及丰富的功能特性，使其成为现代Python开发中不可或缺的工具。无论你是在构建Web API、数据处理管道还是配置管理系统，Pydantic都能帮你写出更安全、更清晰的代码。

引言