Python Pydantic库使用教程:从入门到精通

文章目录

  • Pydantic 完全指南:从入门到实战
  • 一、Pydantic 简介
  • 1.1 什么是 Pydantic?
  • 1.2 核心功能
  • 1.3 主要特性
  • 1.4 Pydantic 生态系统
  • 二、安装
  • 三、基础使用
  • 3.1 基本数据模型定义
  • 3.2 数据验证示例
  • 四、高级功能
  • 4.1 自定义验证器
  • 4.1.1 基本验证器
  • 4.1.2 多字段验证
  • 4.2 嵌套模型
  • 4.3 模型配置:定制化模型行为
  • 4.3.1 常用配置选项详解
  • 4.3.2 高级配置技巧
  • 4.3.3 自定义JSON序列化
  • 五、实际应用场景
  • 5.1 Web 开发(FastAPI)
  • 5.2 API 开发
  • 5.3 数据处理
  • 六、方法与配置全解
  • 6.1 模型方法大全
  • 6.1.1 基础模型方法
  • 6.1.2 高级模型方法
  • 6.2 模型配置全解
  • 6.2.1 基础配置选项
  • 6.2.2 环境变量配置
  • 6.2.3 高级配置技巧
  • 6.3 字段配置详解
  • 6.3.1 Field 函数参数
  • 6.3.2 自定义字段验证
  • 6.4 实用方法组合示例
  • 6.4.1 模型更新模式
  • 6.4.2 动态模型创建
  • 6.4.3 模型继承与扩展
  • 6.5 验证器使用技巧
  • 6.5.1 验证器顺序问题
  • 6.5.2 高级验证选项
  • 七、学习资源
  • 7.1 官方文档
  • 7.2 社区支持
  • Pydantic 完全指南:从入门到实战

    一、Pydantic 简介

    1.1 什么是 Pydantic?

    Pydantic 是一个用于数据验证和设置管理的 Python 库,它通过 Python 类型注解来定义数据模型,提供强大的数据验证功能。

    1.2 核心功能

  • 数据验证:确保输入数据符合预定义的类型和结构
  • 序列化:将复杂数据结构转换为 Python 数据类型
  • 错误处理:提供详细的错误信息
  • 配置管理:支持通过环境变量管理配置
  • 1.3 主要特性

  • 基于 Python 类型注解
  • 高性能(核心验证逻辑用 Rust 编写)
  • 支持 JSON Schema 生成
  • 提供严格模式和宽松模式
  • 支持数据类和 TypedDict
  • 允许自定义验证器和序列化器
  • 1.4 Pydantic 生态系统

    FastAPI 深度集成 Pydantic,用于请求/响应模型验证和自动文档生成。

    相关工具

  • Pydantic-SQLAlchemy:ORM 集成
  • Pydantic-Factories:测试数据生成
  • Pydantic-Settings:配置管理

  • 二、安装

    版本不同,方法差异较大。固定该版本来使用

  • pydantic~=2.10.3
  • pydantic-settings~=2.2.1
  • pip install pydantic
    

    三、基础使用

    3.1 基本数据模型定义

    from pydantic import BaseModel
    
    class User(BaseModel):
        id: int
        name: str
    

    3.2 数据验证示例

    user = User(id=1, name="John")
    print(user)  # 输出验证后的模型
    
    try:
        invalid_user = User(id="one", name=123)
    except ValidationError as e:
        print(e)  # 输出验证错误
    

    四、高级功能

    4.1 自定义验证器

    4.1.1 基本验证器
    from pydantic import BaseModel, field_validator
    
    class User(BaseModel):
        name: str
        age: int
        
        @field_validator('name')
        @classmethod
        def name_must_contain_space(cls, v):
            if ' ' not in v:
                raise ValueError('必须包含空格')
            return v.title()
    
    4.1.2 多字段验证
    @field_validator('age')
    @classmethod
    def age_must_be_positive(cls, v):
        if v <= 0:
            raise ValueError('必须是正整数')
        return v
    

    4.2 嵌套模型

    class Address(BaseModel):
        street: str
        city: str
    
    class User(BaseModel):
        name: str
        address: Address
    

    4.3 模型配置:定制化模型行为

    class ConfigModel(BaseModel):
        name: str
        
        class Config:
            allow_mutation = False  # 使实例不可变
            extra = 'forbid'  # 禁止额外字段
    
    4.3.1 常用配置选项详解
    from pydantic import BaseModel, Field
    from datetime import datetime
    from typing import Optional
    
    class SystemSettings(BaseModel):
        """系统配置模型"""
        
        class Config:
            # 基础配置
            allow_mutation = False  # 模型实例不可变
            extra = 'forbid'       # 禁止额外字段
            validate_all = True    # 验证所有字段(即使没有提供值)
            
            # 日期时间处理
            json_encoders = {
                datetime: lambda v: v.strftime('%Y-%m-%d %H:%M:%S')
            }
            
            # 错误信息模板
            error_msg_templates = {
                'value_error.missing': '字段 {field} 是必填项',
                'type_error.integer': '字段 {field} 必须是整数'
            }
        
        hostname: str = Field(..., min_length=3, max_length=63)
        port: int = Field(1024, ge=1024, le=65535)
        last_updated: Optional[datetime] = None
        debug_mode: bool = False
    
    # 使用示例
    try:
        settings = SystemSettings(
            hostname="my-server",
            port=8080,  # 端口不在允许范围内
            extra_field="value"  # 额外字段会触发错误
        )
    except ValidationError as e:
        print(e.json(indent=2))
    
    4.3.2 高级配置技巧
    from pydantic import BaseModel, validator
    import os
    
    class AppConfig(BaseModel):
        """应用配置模型,支持从环境变量加载"""
        
        class Config:
            env_prefix = 'APP_'  # 环境变量前缀
            case_sensitive = False  # 不区分大小写
            allow_population_by_field_name = True  # 允许通过字段名填充
        
        db_host: str = "localhost"
        db_port: int = 5432
        db_user: str
        db_pass: str
        
        @validator('db_pass')
        def validate_password(cls, value):
            if len(value) < 8:
                raise ValueError('数据库密码至少需要8个字符')
            return value
    
    # 设置环境变量
    os.environ['APP_DB_USER'] = 'admin'
    os.environ['APP_DB_PASS'] = 'securepassword123'
    
    # 从环境变量创建配置
    config = AppConfig()
    print("数据库配置:", config.dict())
    
    4.3.3 自定义JSON序列化
    from pydantic import BaseModel
    from decimal import Decimal
    from datetime import date
    
    class FinancialRecord(BaseModel):
        transaction_id: str
        amount: Decimal
        date: date
        description: str = None
        
        class Config:
            json_encoders = {
                Decimal: lambda v: str(round(v, 2)),
                date: lambda v: v.isoformat()
            }
            
            # 控制dict输出
            @staticmethod
            def schema_extra(schema, model):
                schema['example'] = {
                    "transaction_id": "txn_12345",
                    "amount": "100.00",
                    "date": "2023-01-01",
                    "description": "Sample transaction"
                }
    
    record = FinancialRecord(
        transaction_id="txn_98765",
        amount=Decimal("1234.5678"),
        date=date.today()
    )
    
    print("JSON输出:", record.json(indent=2))
    print("字典输出:", record.dict())
    

    五、实际应用场景

    5.1 Web 开发(FastAPI)

    from fastapi import FastAPI
    from pydantic import BaseModel
    
    app = FastAPI()
    
    class Item(BaseModel):
        name: str
        price: float
    
    @app.post("/items/")
    async def create_item(item: Item):
        return item
    

    5.2 API 开发

    class UserResponse(BaseModel):
        id: int
        name: str
        email: str
    
    def get_user() -> UserResponse:
        return UserResponse(id=1, name="Alice", email="alice@example.com")
    

    5.3 数据处理

    def process_data(raw_data: dict) -> User:
        try:
            return User(**raw_data)
        except ValidationError as e:
            print(f"数据验证失败: {e}")
            raise
    

    六、方法与配置全解

    6.1 模型方法大全

  • model_dump()
  • user = User(id=1, name="John")
    print(user.model_dump())  # 输出: {'id': 1, 'name': 'John'}
    
  • model_json_schema()
  • print(User.model_json_schema())  # 输出模型的JSON Schema
    
    6.1.1 基础模型方法
    from pydantic import BaseModel
    from datetime import datetime
    
    class User(BaseModel):
        id: int
        name: str
        created_at: datetime = None
    
    # 1. 构造方法
    user = User(id=1, name="Alice")  # 标准构造
    user = User.parse_obj({"id": 2, "name": "Bob"})  # 从字典构造
    
    # 2. 序列化方法
    print(user.dict())  # 转为字典
    # 输出: {'id': 1, 'name': 'Alice', 'created_at': None}
    
    print(user.json())  # 转为JSON字符串
    # 输出: {"id": 1, "name": "Alice", "created_at": null}
    
    # 3. 拷贝方法
    user_copy = user.copy()  # 浅拷贝
    user_deepcopy = user.copy(deep=True)  # 深拷贝
    
    # 4. 字段访问方法
    print(user.dict(exclude={'created_at'}))  # 排除特定字段
    print(user.dict(include={'id'}))  # 只包含特定字段
    
    6.1.2 高级模型方法
    # 1. 模式导出
    print(User.schema())  # 获取JSON Schema
    print(User.schema_json(indent=2))  # 获取格式化的JSON Schema
    
    # 2. 字段信息获取
    print(User.__fields__)  # 获取所有字段定义
    print(User.__fields__['id'].type_)  # 获取字段类型(int)
    
    # 3. 更新方法
    updated_user = user.copy(update={"name": "Alice Smith"})
    
    # 4. 解析原始数据
    raw_data = '{"id": 3, "name": "Carol"}'
    user = User.parse_raw(raw_data)  # 从JSON字符串解析
    

    6.2 模型配置全解

    6.2.1 基础配置选项
    class ConfigDemo(BaseModel):
        name: str
        age: int
        
        class Config:
            # 字段处理配置
            allow_mutation = False  # 禁止修改模型实例
            extra = 'forbid'  # 处理额外字段: 'allow'|'ignore'|'forbid'
            underscore_attrs_are_private = True  # 下划线开头属性为私有
            
            # 验证配置
            validate_all = True  # 验证所有字段(包括可选字段)
            validate_assignment = True  # 赋值时验证
            
            # 序列化配置
            json_encoders = {
                datetime: lambda v: v.isoformat()  # 自定义JSON编码
            }
            json_loads = lambda x: x  # 自定义JSON解析
            json_dumps = lambda x, **kwargs: x  # 自定义JSON序列化
    
    6.2.2 环境变量配置
    import os
    # 在我使用的v2版本中已经弃用
    # from pydantic import BaseSettings
    from pydantic_settings import BaseSettings
    
    class AppSettings(BaseSettings):
        # 自动从环境变量读取(不区分大小写)
        db_host: str = 'localhost'
        db_port: int = 5432
        
        class Config:
            env_prefix = 'APP_'  # 环境变量前缀(APP_DB_HOST)
            case_sensitive = False  # 不区分大小写
            env_file = '.env'  # 从.env文件加载
            env_file_encoding = 'utf-8'
    
    os.environ['APP_DB_HOST'] = 'prod.db.example.com'
    settings = AppSettings()
    
    # 在我使用的v2版本中已经弃用
    # print(settings.dict())
    print(settings.model_dump())
    

    输出:

    {'db_host': 'prod.db.example.com', 'db_port': 5432}
    
    6.2.3 高级配置技巧
    from typing import Any
    
    from pydantic import BaseModel
    from pydantic import Field
    
    class AdvancedConfig(BaseModel):
        value: Any = Field(
            alias='VALUE',
            title='配置值',
            description='可以是任何类型的值'
        )
    
        class Config:
            # 自定义schema额外信息
            @staticmethod
            def json_schema_extra(schema: dict, model) -> None:
                schema['examples'] = [{'VALUE': 42}]
    
    
    # 使用别名创建实例
    config = AdvancedConfig.model_validate({'VALUE': 'example'})
    print(config.model_dump_json(indent=2))
    

    输出:

    {
      "value": "example"
    }
    

    6.3 字段配置详解

    6.3.1 Field 函数参数
    import json
    from pydantic import BaseModel, Field
    from typing import Optional
    
    
    class Product(BaseModel):
        # 基础字段配置
        id: int = Field(..., gt=0, description="产品ID必须为正整数")
        name: str = Field(
            ...,
            min_length=2,
            max_length=100,
            pattern=r'^[a-zA-Z0-9_\- ]+$',
            title="产品名称",
            description="2-100个字符,允许字母、数字、空格、下划线和连字符"
        )
    
        # 可选字段配置
        price: Optional[float] = Field(
            None,
            gt=0,
            alias="productPrice",
            example=99.99
        )
    
        # 复杂配置
        tags: list[str] = Field(
            default_factory=list,
            min_items=1,
            max_items=10,
            description="产品标签列表"
        )
    
    
    # 使用示例
    product = Product(
        id=1,
        name="Premium Product",
        price=199.99,
        tags=["new", "featured"]
    )
    print(json.dumps(product.model_json_schema(), indent=2))
    
    6.3.2 自定义字段验证
    from pydantic import validator
    
    class CustomFieldDemo(BaseModel):
        username: str
        password: str = Field(..., min_length=8)
        
        @validator('username')
        def username_alphanumeric(cls, v):
            if not v.isalnum():
                raise ValueError('必须是字母数字组合')
            return v
        
        @validator('password')
        def password_complexity(cls, v):
            if not any(c.isupper() for c in v):
                raise ValueError('必须包含大写字母')
            if not any(c.isdigit() for c in v):
                raise ValueError('必须包含数字')
            return v
        
        class Config:
            fields = {
                'password': {'exclude': True}  # 从dict()/json()输出中排除
            }
    
    # 使用示例
    user = CustomFieldDemo(username="Alice123", password="Pass1234")
    print(user.dict())  # 密码字段被排除
    

    6.4 实用方法组合示例

    6.4.1 模型更新模式
    class UpdateModel(BaseModel):
        name: str = None
        age: int = None
        
        class Config:
            extra = 'forbid'
            validate_assignment = True
        
        def apply_update(self, **kwargs):
            """安全更新方法"""
            for field, value in kwargs.items():
                if field in self.__fields__:
                    setattr(self, field, value)
    
    user = UpdateModel()
    user.apply_update(name="Alice", age=30, invalid="value")  # 忽略无效字段
    print(user.dict())
    
    6.4.2 动态模型创建
    from pydantic import create_model
    
    DynamicModel = create_model(
        'DynamicModel',
        id=(int, Field(..., gt=0)),
        name=(str, Field(..., max_length=100)),
        __config__=dict(extra='forbid')
    )
    
    instance = DynamicModel(id=1, name="Test")
    print(instance)
    
    6.4.3 模型继承与扩展
    class BaseUser(BaseModel):
        id: int
        username: str
    
    class AdminUser(BaseUser):
        permissions: list[str]
        is_active: bool = True
        
        class Config:
            allow_mutation = False
    
    admin = AdminUser(
        id=1,
        username="admin",
        permissions=["read", "write", "delete"]
    )
    print(admin.json(indent=2))
    

    6.5 验证器使用技巧

    6.5.1 验证器顺序问题
    # 正确顺序
    @field_validator('field')
    @classmethod
    def validator(cls, v):
        return v
    
    # 错误顺序(可能失效)
    @classmethod
    @field_validator('field')
    def validator(cls, v):
        return v
    
    6.5.2 高级验证选项
    @field_validator('field', mode='before')  # 在基本验证前执行
    @field_validator('field', check_fields=False)  # 不检查字段是否存在
    

    七、学习资源

    7.1 官方文档

    Pydantic 官方文档 是最全面的学习资源。

    7.2 社区支持

  • GitHub 仓库
  • Stack Overflow 上的 Pydantic 标签
  • Python 相关论坛和社区

  • 作者:Hellchin

    物联沃分享整理
    物联沃-IOTWORD物联网 » Python Pydantic库使用教程:从入门到精通

    发表回复