导航菜单

  • 1.vector
  • 2.milvus
  • 3.pymilvus
  • 4.rag
  • 5.rag_measure
  • 7.search
  • ragflow
  • heapq
  • HNSW
  • cosine_similarity
  • math
  • typing
  • etcd
  • minio
  • collections
  • jieba
  • random
  • beautifulsoup4
  • chromadb
  • sentence_transformers
  • numpy
  • lxml
  • openpyxl
  • PyMuPDF
  • python-docx
  • requests
  • python-pptx
  • text_splitter
  • all-MiniLM-L6-v2
  • openai
  • llm
  • BPETokenizer
  • Flask
  • RAGAS
  • BagofWords
  • langchain
  • Pydantic
  • abc
  • faiss
  • MMR
  • scikit-learn
  • 1. 什么是 Pydantic?
    • 1.1 简单理解
    • 1.2 核心特点
    • 1.3 类比理解
  • 2. 为什么需要 Pydantic?
    • 2.1 传统方式的问题
    • 2.2 Pydantic 的优势
  • 3. 前置知识
    • 3.1 Python 类型注解(Type Hints)
      • 3.1.1 基本类型注解
      • 3.1.2 复杂类型注解
    • 3.2 Python 类的基础知识
      • 3.2.1 定义类
      • 3.2.2 类的继承
    • 3.3 字典解包(**kwargs)
    • 3.4 装饰器(Decorator)
  • 4. 安装 Pydantic
    • 4.1 使用 pip 安装
      • 4.1.1 Windows 系统
      • 4.1.2 Mac 系统
    • 4.2 验证安装
  • 5. 快速开始
    • 5.1 用户信息验证
    • 5.2 运行代码
    • 5.3 运行结果
    • 5.4 关键点理解
  • 6. 核心功能
    • 6.1 数据验证
      • 6.1.1 类型验证
      • 6.1.2 运行结果
    • 6.2 数据序列化
    • 6.3 字段验证器
    • 6.4 字段配置(Field)
    • 6.5 嵌套模型
  • 7. 高级功能
    • 7.1 可选字段和默认值
    • 7.2 数据转换和清洗
  • 8. 实际应用场景
    • 8.1 API 数据验证(FastAPI 示例)
    • 8.2 配置文件管理
    • 8.3 数据验证和清洗工具
  • 9. 常见问题
    • 9.1 Q1: Pydantic 和普通 Python 类有什么区别?
    • 9.2 Q2: 如何查看模型的字段信息?
    • 9.3 Q3: 如何处理日期时间?
    • 9.4 Q4: 如何只验证部分字段?
    • 9.5 Q5: Pydantic 性能如何?
  • 10. 总结

1. 什么是 Pydantic? #

1.1 简单理解 #

Pydantic 是一个 Python 库,它的主要作用是验证数据是否正确。想象一下,你有一个表单,用户需要填写姓名、年龄、邮箱等信息。Pydantic 就像一个严格的检查员,确保用户输入的数据符合要求:

  • 年龄必须是数字,不能是文字
  • 邮箱必须包含 @ 符号
  • 姓名不能为空

如果数据不符合要求,Pydantic 会立即告诉你哪里出错了。

1.2 核心特点 #

  1. 基于类型注解:使用 Python 的类型提示来定义数据应该是什么样子
  2. 自动验证:创建对象时自动检查数据是否正确
  3. 类型转换:如果可能,会自动将数据转换为正确的类型(比如把字符串 "123" 转换为数字 123)
  4. 易于使用:代码简洁,容易理解

1.3 类比理解 #

把 Pydantic 想象成一个智能的表格模板:

  • 你定义好表格的格式(哪些列是必需的,哪些列是什么类型)
  • 当有人填写表格时,Pydantic 自动检查:
    • 必填项是否都填了?
    • 数字列填的是数字吗?
    • 邮箱格式正确吗?
  • 如果填写错误,立即提示问题在哪里

2. 为什么需要 Pydantic? #

2.1 传统方式的问题 #

在 Python 中,如果不使用 Pydantic,验证数据通常需要写很多 if-else 判断:

# 传统方式:手动验证数据
def create_user(user_data):
    # 检查 id 是否存在
    if 'id' not in user_data:
        raise ValueError("缺少 id")

    # 检查 id 是否是整数
    if not isinstance(user_data['id'], int):
        raise ValueError("id 必须是整数")

    # 检查 name 是否存在
    if 'name' not in user_data:
        raise ValueError("缺少 name")

    # 检查 name 是否是字符串
    if not isinstance(user_data['name'], str):
        raise ValueError("name 必须是字符串")

    # ... 还有很多检查 ...

    return user_data

这种方式的问题:

  • 代码冗长:需要写很多重复的检查代码
  • 容易出错:可能忘记检查某些字段
  • 难以维护:当数据结构改变时,需要修改很多地方

2.2 Pydantic 的优势 #

使用 Pydantic,同样的功能只需要几行代码:

from pydantic import BaseModel

# 定义数据模型
class User(BaseModel):
    id: int
    name: str

# 自动验证
user = User(id=1, name="张三")  # 自动检查类型

优势:

  • 代码简洁:定义一次,自动验证
  • 自动类型检查:不需要手动写 if-else
  • 清晰的错误信息:如果数据错误,会告诉你具体哪里错了
  • 易于维护:修改模型定义即可

3. 前置知识 #

在学习 Pydantic 之前,你需要了解一些 Python 基础知识。如果你已经熟悉这些内容,可以跳过。

3.1 Python 类型注解(Type Hints) #

类型注解是 Python 3.5+ 引入的功能,用来标注变量、函数参数和返回值的类型。

3.1.1 基本类型注解 #

# 标注变量的类型
name: str = "张三"        # name 是字符串类型
age: int = 25            # age 是整数类型
price: float = 99.99     # price 是浮点数类型
is_active: bool = True   # is_active 是布尔类型

# 标注函数参数和返回值的类型
def add(a: int, b: int) -> int:
    """
    函数 add 接收两个整数参数 a 和 b,返回一个整数
    """
    return a + b

result = add(3, 5)  # result 是 8

3.1.2 复杂类型注解 #

from typing import List, Dict, Optional

# List[str] 表示字符串列表
names: List[str] = ["张三", "李四", "王五"]

# Dict[str, int] 表示字典,键是字符串,值是整数
scores: Dict[str, int] = {"张三": 90, "李四": 85}

# Optional[str] 表示可以是字符串,也可以是 None
email: Optional[str] = None  # 可以是字符串,也可以是 None
email = "zhangsan@example.com"  # 也可以赋值字符串

注意:类型注解不会影响程序运行,只是给开发者和工具(如 Pydantic)提供信息。

3.2 Python 类的基础知识 #

3.2.1 定义类 #

# 定义一个简单的类
class Person:
    # __init__ 是构造函数,创建对象时自动调用
    def __init__(self, name: str, age: int):
        # self 表示对象本身
        self.name = name  # 设置对象的 name 属性
        self.age = age    # 设置对象的 age 属性

    # 定义方法
    def introduce(self):
        return f"我是{self.name},今年{self.age}岁"

# 创建对象
person = Person("张三", 25)
print(person.introduce())  # 输出: 我是张三,今年25岁

3.2.2 类的继承 #

# 父类(基类)
class Animal:
    def __init__(self, name: str):
        self.name = name

    def speak(self):
        return "动物在叫"

# 子类(派生类),继承自 Animal
class Dog(Animal):
    def speak(self):
        return f"{self.name}在汪汪叫"

# 创建子类对象
dog = Dog("旺财")
print(dog.speak())  # 输出: 旺财在汪汪叫

Pydantic 中的 BaseModel 就是一个基类,我们定义的模型类继承自它,就自动获得了数据验证等功能。

3.3 字典解包(**kwargs) #

在 Python 中,** 用于解包字典。

# 定义一个字典
user_data = {"id": 1, "name": "张三", "age": 25}

# 传统方式:逐个传递参数
def create_user(id, name, age):
    return {"id": id, "name": name, "age": age}

user1 = create_user(user_data["id"], user_data["name"], user_data["age"])

# 使用 ** 解包字典,自动传递所有键值对
user2 = create_user(**user_data)  # 等同于 create_user(id=1, name="张三", age=25)

在 Pydantic 中,我们经常用 ** 从字典创建对象:

user = User(**user_data)  # 从字典创建 User 对象

3.4 装饰器(Decorator) #

装饰器是 Python 的高级特性,用于修改函数或类的行为。

# 定义一个装饰器
def my_decorator(func):
    def wrapper(*args, **kwargs):
        print("函数执行前")
        result = func(*args, **kwargs)
        print("函数执行后")
        return result
    return wrapper

# 使用装饰器
@my_decorator
def say_hello():
    print("Hello!")

say_hello()
# 输出:
# 函数执行前
# Hello!
# 函数执行后

在 Pydantic 中,@validator 是一个装饰器,用于定义字段验证函数。

4. 安装 Pydantic #

4.1 使用 pip 安装 #

4.1.1 Windows 系统 #

打开命令提示符(CMD)或 PowerShell:

# 安装 Pydantic
pip install pydantic

如果使用 Python 3,可能需要使用 pip3:

pip3 install pydantic

4.1.2 Mac 系统 #

打开终端(Terminal):

# 安装 Pydantic
pip3 install pydantic

4.2 验证安装 #

创建一个测试文件 test_pydantic.py:

# 导入 pydantic,检查是否安装成功
try:
    from pydantic import BaseModel
    print("Pydantic 安装成功!")
except ImportError:
    print("Pydantic 未安装,请运行: pip install pydantic")

运行测试:

Windows:

python test_pydantic.py

Mac:

python3 test_pydantic.py

如果看到 "Pydantic 安装成功!",说明安装正确。

5. 快速开始 #

让我们通过一个简单的例子快速了解 Pydantic 的基本用法。

5.1 用户信息验证 #

创建一个文件 example_01_basic.py:

# ============================================
# 第一个 Pydantic 例子:用户信息验证
# ============================================

# 导入 BaseModel,这是 Pydantic 的基础类
from pydantic import BaseModel

# 定义一个用户模型
# 继承 BaseModel 后,这个类就自动具有数据验证功能
class User(BaseModel):
    # id 字段必须是整数类型
    id: int
    # name 字段必须是字符串类型
    name: str
    # email 字段必须是字符串类型
    email: str
    # age 字段是整数类型,默认值是 18
    # 如果创建对象时不提供 age,会自动使用默认值
    age: int = 18

# 示例1:创建用户对象(所有数据都正确)
print("=" * 50)
print("示例1:创建用户对象(数据正确)")
print("=" * 50)
user1 = User(id=1, name="张三", email="zhangsan@example.com", age=25)
print(f"用户对象: {user1}")
print(f"用户姓名: {user1.name}")
print(f"用户年龄: {user1.age}")

# 示例2:使用默认值(不提供 age)
print("\n" + "=" * 50)
print("示例2:使用默认值")
print("=" * 50)
user2 = User(id=2, name="李四", email="lisi@example.com")
print(f"用户对象: {user2}")
print(f"用户年龄(使用默认值): {user2.age}")

# 示例3:从字典创建对象
print("\n" + "=" * 50)
print("示例3:从字典创建对象")
print("=" * 50)
user_data = {
    "id": 3,
    "name": "王五",
    "email": "wangwu@example.com",
    "age": 30
}
# 使用 ** 解包字典,将字典的键值对作为参数传递
user3 = User(**user_data)
print(f"用户对象: {user3}")

# 示例4:数据验证(错误的数据)
print("\n" + "=" * 50)
print("示例4:数据验证(错误的数据)")
print("=" * 50)
try:
    # 尝试用字符串作为 id(应该是整数)
    user4 = User(id="不是数字", name="赵六", email="zhaoliu@example.com")
except Exception as e:
    # 捕获异常并打印错误信息
    print(f"验证错误: {e}")
    print("\nPydantic 自动检测到 id 应该是整数,但提供了字符串!")

# 示例5:缺少必需字段
print("\n" + "=" * 50)
print("示例5:缺少必需字段")
print("=" * 50)
try:
    # 缺少必需的 email 字段
    user5 = User(id=5, name="孙七")
except Exception as e:
    print(f"验证错误: {e}")
    print("\nPydantic 检测到缺少必需的 email 字段!")

5.2 运行代码 #

Windows:

python example_01_basic.py

Mac:

python3 example_01_basic.py

5.3 运行结果 #

==================================================
示例1:创建用户对象(数据正确)
==================================================
用户对象: id=1 name='张三' email='zhangsan@example.com' age=25
用户姓名: 张三
用户年龄: 25

==================================================
示例2:使用默认值
==================================================
用户对象: id=2 name='李四' email='lisi@example.com' age=18
用户年龄(使用默认值): 18

==================================================
示例3:从字典创建对象
==================================================
用户对象: id=3 name='王五' email='wangwu@example.com' age=30

==================================================
示例4:数据验证(错误的数据)
==================================================
验证错误: 1 validation error for User
id
  value is not a valid integer (type=type_error.integer)

Pydantic 自动检测到 id 应该是整数,但提供了字符串!

==================================================
示例5:缺少必需字段
==================================================
验证错误: 1 validation error for User
email
  field required (type=value_error.missing)

Pydantic 检测到缺少必需的 email 字段!

5.4 关键点理解 #

  1. 定义模型:继承 BaseModel 并定义字段类型
  2. 自动验证:创建对象时自动检查数据类型
  3. 默认值:可以为字段设置默认值
  4. 错误提示:数据错误时,Pydantic 会给出清晰的错误信息

6. 核心功能 #

6.1 数据验证 #

Pydantic 的核心功能是自动验证数据。让我们详细看看它是如何工作的。

6.1.1 类型验证 #

创建文件 example_02_validation.py:

# ============================================
# 数据验证
# ============================================

from pydantic import BaseModel, ValidationError

# 定义一个产品模型
class Product(BaseModel):
    # 产品名称必须是字符串
    name: str
    # 价格必须是浮点数
    price: float
    # 库存数量必须是整数
    stock: int

# 示例1:正确的数据
print("=" * 50)
print("示例1:正确的数据")
print("=" * 50)
product1 = Product(name="笔记本电脑", price=5999.99, stock=10)
print(f"产品: {product1}")
print(f"产品名称类型: {type(product1.name)}")
print(f"产品价格类型: {type(product1.price)}")

# 示例2:自动类型转换
print("\n" + "=" * 50)
print("示例2:自动类型转换")
print("=" * 50)
# 注意:price 传入的是字符串 "5999.99",但 Pydantic 会自动转换为浮点数
# stock 传入的是字符串 "10",但 Pydantic 会自动转换为整数
product2 = Product(name="鼠标", price="29.99", stock="5")
print(f"产品: {product2}")
print(f"价格类型(已自动转换): {type(product2.price)}")
print(f"库存类型(已自动转换): {type(product2.stock)}")

# 示例3:无法转换的类型错误
print("\n" + "=" * 50)
print("示例3:无法转换的类型错误")
print("=" * 50)
try:
    # 尝试将 "不是数字" 转换为整数,会失败
    product3 = Product(name="键盘", price=99.99, stock="不是数字")
except ValidationError as e:
    # ValidationError 是 Pydantic 的验证错误类型
    print("验证失败!")
    print(f"错误详情: {e}")
    # 可以获取更详细的错误信息
    print("\n详细错误列表:")
    for error in e.errors():
        print(f"  字段: {error['loc']}")
        print(f"  错误类型: {error['type']}")
        print(f"  错误信息: {error['msg']}")

# 示例4:缺少必需字段
print("\n" + "=" * 50)
print("示例4:缺少必需字段")
print("=" * 50)
try:
    # 缺少必需的 stock 字段
    product4 = Product(name="显示器", price=1999.99)
except ValidationError as e:
    print("验证失败!缺少必需字段")
    print(f"错误: {e}")

# 示例5:使用 Optional 定义可选字段
print("\n" + "=" * 50)
print("示例5:可选字段")
print("=" * 50)
from typing import Optional

# 定义一个新模型,description 是可选的
class ProductWithDescription(BaseModel):
    name: str
    price: float
    # Optional[str] 表示可以是字符串,也可以是 None
    # = None 表示默认值是 None
    description: Optional[str] = None

# 不提供 description(使用默认值 None)
product5 = ProductWithDescription(name="耳机", price=299.99)
print(f"产品(无描述): {product5}")

# 提供 description
product6 = ProductWithDescription(name="音响", price=599.99, description="高品质蓝牙音响")
print(f"产品(有描述): {product6}")

6.1.2 运行结果 #

运行 python example_02_validation.py(Windows)或 python3 example_02_validation.py(Mac),你会看到详细的验证过程。

6.2 数据序列化 #

序列化是将对象转换为可以存储或传输的格式(如字典、JSON)。反序列化是相反的过程。

创建文件 example_03_serialization.py:

# ============================================
# 数据序列化和反序列化
# ============================================

from pydantic import BaseModel
import json

# 定义用户模型
class User(BaseModel):
    id: int
    name: str
    email: str
    age: int = 18

# 创建一个用户对象
user = User(id=1, name="张三", email="zhangsan@example.com", age=25)

print("=" * 50)
print("原始对象")
print("=" * 50)
print(f"用户对象: {user}")
print(f"对象类型: {type(user)}")

# 方法1:转换为字典
print("\n" + "=" * 50)
print("方法1:转换为字典(dict())")
print("=" * 50)
# dict() 方法将 Pydantic 对象转换为普通字典
user_dict = user.dict()
print(f"字典: {user_dict}")
print(f"字典类型: {type(user_dict)}")
print(f"可以像普通字典一样访问: {user_dict['name']}")

# 方法2:转换为 JSON 字符串
print("\n" + "=" * 50)
print("方法2:转换为 JSON 字符串(json())")
print("=" * 50)
# json() 方法将对象转换为 JSON 格式的字符串
user_json = user.json()
print(f"JSON 字符串: {user_json}")
print(f"JSON 类型: {type(user_json)}")
# JSON 字符串可以保存到文件或通过网络传输

# 方法3:转换为格式化的 JSON(更易读)
print("\n" + "=" * 50)
print("方法3:格式化的 JSON")
print("=" * 50)
# 使用 json.dumps 格式化输出
user_json_pretty = user.json(indent=2)
print(user_json_pretty)

# 方法4:从字典创建对象(反序列化)
print("\n" + "=" * 50)
print("方法4:从字典创建对象")
print("=" * 50)
# 假设我们从 API 或数据库获取了字典数据
data_from_api = {
    "id": 2,
    "name": "李四",
    "email": "lisi@example.com",
    "age": 30
}
# 使用 ** 解包字典创建对象
user_from_dict = User(**data_from_api)
print(f"从字典创建的对象: {user_from_dict}")

# 方法5:从 JSON 字符串创建对象
print("\n" + "=" * 50)
print("方法5:从 JSON 字符串创建对象")
print("=" * 50)
# 假设我们收到 JSON 字符串(比如从网络请求)
json_string = '{"id": 3, "name": "王五", "email": "wangwu@example.com", "age": 28}'
# parse_raw() 方法从 JSON 字符串创建对象
user_from_json = User.parse_raw(json_string)
print(f"从 JSON 创建的对象: {user_from_json}")

# 方法6:从文件读取 JSON 并创建对象
print("\n" + "=" * 50)
print("方法6:从文件读取(示例)")
print("=" * 50)
# 首先创建一个 JSON 文件
json_data = {"id": 4, "name": "赵六", "email": "zhaoliu@example.com", "age": 35}
with open("user_data.json", "w", encoding="utf-8") as f:
    json.dump(json_data, f, ensure_ascii=False, indent=2)
print("已创建 user_data.json 文件")

# 从文件读取并创建对象
with open("user_data.json", "r", encoding="utf-8") as f:
    data = json.load(f)
    user_from_file = User(**data)
    print(f"从文件创建的对象: {user_from_file}")

# 实际应用场景:API 数据交换
print("\n" + "=" * 50)
print("实际应用:模拟 API 数据交换")
print("=" * 50)
# 模拟:客户端发送数据到服务器
client_data = {"id": 5, "name": "孙七", "email": "sunqi@example.com"}
print(f"客户端发送: {client_data}")

# 服务器接收并验证
server_user = User(**client_data)
print(f"服务器验证通过: {server_user}")

# 服务器处理后将对象转换为 JSON 返回
response_json = server_user.json()
print(f"服务器返回 JSON: {response_json}")

6.3 字段验证器 #

除了类型验证,我们还可以自定义验证规则。

创建文件 example_04_validator.py:

# ============================================
# 自定义字段验证器
# ============================================

from pydantic import BaseModel, validator, ValidationError

# 定义商品模型,包含自定义验证
class Product(BaseModel):
    # 商品名称
    name: str
    # 商品价格
    price: float
    # 商品库存
    stock: int

    # 使用 @validator 装饰器定义价格验证器
    # 'price' 表示验证 price 字段
    @validator('price')
    def price_must_be_positive(cls, v):
        """
        验证价格必须大于 0
        cls: 类本身(Product)
        v: 字段的值(price 的值)
        """
        # 如果价格小于等于 0,抛出错误
        if v <= 0:
            raise ValueError('价格必须大于 0')
        # 如果验证通过,返回原值(也可以返回修改后的值)
        return v

    # 验证库存
    @validator('stock')
    def stock_must_be_non_negative(cls, v):
        """
        验证库存不能为负数
        """
        if v < 0:
            raise ValueError('库存不能为负数')
        return v

    # 验证商品名称
    @validator('name')
    def name_must_not_be_empty(cls, v):
        """
        验证商品名称不能为空
        """
        # strip() 去除首尾空格
        v = v.strip()
        if not v:  # 如果去除空格后为空
            raise ValueError('商品名称不能为空')
        return v  # 返回去除空格后的值

# 示例1:正确的数据
print("=" * 50)
print("示例1:正确的数据")
print("=" * 50)
product1 = Product(name="笔记本电脑", price=5999.99, stock=10)
print(f"商品: {product1}")
print("验证通过!")

# 示例2:价格验证失败
print("\n" + "=" * 50)
print("示例2:价格验证失败")
print("=" * 50)
try:
    # 价格是负数,验证会失败
    product2 = Product(name="鼠标", price=-10.0, stock=5)
except ValidationError as e:
    print("验证失败!")
    for error in e.errors():
        print(f"  字段: {error['loc']}")
        print(f"  错误: {error['msg']}")

# 示例3:库存验证失败
print("\n" + "=" * 50)
print("示例3:库存验证失败")
print("=" * 50)
try:
    # 库存是负数,验证会失败
    product3 = Product(name="键盘", price=99.99, stock=-5)
except ValidationError as e:
    print("验证失败!")
    for error in e.errors():
        print(f"  字段: {error['loc']}")
        print(f"  错误: {error['msg']}")

# 示例4:名称验证(自动去除空格)
print("\n" + "=" * 50)
print("示例4:名称验证(自动处理)")
print("=" * 50)
# 名称前后有空格,验证器会自动去除
product4 = Product(name="  显示器  ", price=1999.99, stock=8)
print(f"商品名称(已去除空格): '{product4.name}'")

# 示例5:多个验证器同时工作
print("\n" + "=" * 50)
print("示例5:多个验证器")
print("=" * 50)
try:
    # 多个字段都有问题
    product5 = Product(name="", price=-100, stock=-10)
except ValidationError as e:
    print("验证失败!发现多个错误:")
    for error in e.errors():
        print(f"  {error['loc']}: {error['msg']}")

6.4 字段配置(Field) #

使用 Field 可以更灵活地配置字段。

创建文件 example_05_field.py:

# ============================================
# 字段配置(Field)
# ============================================

from pydantic import BaseModel, Field
from typing import Optional

# 定义商品模型,使用 Field 配置字段
class Item(BaseModel):
    # Field 用于配置字段的详细属性
    # ... 表示该字段是必需的(不能省略)
    # min_length=1 表示最小长度为 1
    # max_length=50 表示最大长度为 50
    # description 是字段的描述信息
    name: str = Field(..., min_length=1, max_length=50, description="商品名称")

    # gt=0 表示必须大于 0(greater than)
    # description 提供字段说明
    price: float = Field(..., gt=0, description="商品价格,必须大于0")

    # Optional[str] 表示可以是字符串或 None
    # None 是默认值
    # max_length=200 限制最大长度
    description: Optional[str] = Field(None, max_length=200, description="商品描述")

    # ge=0 表示大于等于 0(greater than or equal)
    # le=100 表示小于等于 100(less than or equal)
    # 10.0 是默认值
    discount: float = Field(10.0, ge=0, le=100, description="折扣百分比,0-100之间")

# 示例1:正确的数据
print("=" * 50)
print("示例1:正确的数据")
print("=" * 50)
item1 = Item(name="笔记本电脑", price=5999.99, description="高性能游戏本", discount=15.0)
print(f"商品: {item1}")

# 示例2:使用默认值
print("\n" + "=" * 50)
print("示例2:使用默认值")
print("=" * 50)
item2 = Item(name="鼠标", price=29.99)
print(f"商品(使用默认折扣): {item2}")

# 示例3:字段长度验证
print("\n" + "=" * 50)
print("示例3:字段长度验证")
print("=" * 50)
from pydantic import ValidationError

try:
    # 名称太长,超过 50 个字符
    long_name = "a" * 51  # 创建 51 个字符的字符串
    item3 = Item(name=long_name, price=99.99)
except ValidationError as e:
    print("验证失败!")
    for error in e.errors():
        print(f"  {error['loc']}: {error['msg']}")

# 示例4:数值范围验证
print("\n" + "=" * 50)
print("示例4:数值范围验证")
print("=" * 50)
try:
    # 折扣超过 100,验证失败
    item4 = Item(name="键盘", price=99.99, discount=150.0)
except ValidationError as e:
    print("验证失败!")
    for error in e.errors():
        print(f"  {error['loc']}: {error['msg']}")

# 示例5:查看字段信息
print("\n" + "=" * 50)
print("示例5:查看模型结构")
print("=" * 50)
# schema() 方法返回模型的完整结构信息
schema = Item.schema()
print("模型结构(JSON Schema):")
import json
print(json.dumps(schema, indent=2, ensure_ascii=False))

6.5 嵌套模型 #

一个模型可以包含另一个模型,形成嵌套结构。

创建文件 example_06_nested.py:

# ============================================
# 嵌套模型
# ============================================

from pydantic import BaseModel
from typing import List, Optional

# 定义地址模型
class Address(BaseModel):
    # 街道地址
    street: str
    # 城市
    city: str
    # 邮政编码
    zip_code: str

# 定义公司模型,包含地址
class Company(BaseModel):
    # 公司名称
    name: str
    # 公司地址,类型是 Address(嵌套模型)
    address: Address

# 定义员工模型
class Employee(BaseModel):
    # 员工姓名
    name: str
    # 员工年龄
    age: int
    # 所属公司(嵌套模型)
    company: Company
    # 技能列表(List[str] 表示字符串列表)
    skills: List[str] = []

# 示例1:创建嵌套对象
print("=" * 50)
print("示例1:创建嵌套对象")
print("=" * 50)
# 方式1:先创建 Address 对象,再创建 Company 对象
address = Address(street="科技路123号", city="北京", zip_code="100000")
company = Company(name="ABC公司", address=address)
employee = Employee(name="张三", age=25, company=company, skills=["Python", "Java"])
print(f"员工: {employee}")
print(f"员工公司: {employee.company.name}")
print(f"公司地址: {employee.company.address.city}")

# 示例2:使用字典创建(更常用)
print("\n" + "=" * 50)
print("示例2:使用字典创建")
print("=" * 50)
# 可以直接用字典创建,Pydantic 会自动处理嵌套
employee_data = {
    "name": "李四",
    "age": 30,
    "company": {
        "name": "XYZ公司",
        "address": {
            "street": "创新路456号",
            "city": "上海",
            "zip_code": "200000"
        }
    },
    "skills": ["JavaScript", "React", "Node.js"]
}
employee2 = Employee(**employee_data)
print(f"员工: {employee2}")
print(f"技能列表: {employee2.skills}")

# 示例3:嵌套验证
print("\n" + "=" * 50)
print("示例3:嵌套验证")
print("=" * 50)
from pydantic import ValidationError

try:
    # 地址信息不完整(缺少 zip_code)
    employee3 = Employee(
        name="王五",
        age=28,
        company={
            "name": "DEF公司",
            "address": {
                "street": "商业街789号",
                "city": "广州"
                # 缺少 zip_code
            }
        }
    )
except ValidationError as e:
    print("验证失败!嵌套模型验证出错:")
    for error in e.errors():
        print(f"  {error['loc']}: {error['msg']}")

# 示例4:复杂嵌套(列表中的模型)
print("\n" + "=" * 50)
print("示例4:列表中的嵌套模型")
print("=" * 50)

# 定义订单项模型
class OrderItem(BaseModel):
    product_name: str
    quantity: int
    price: float

# 定义订单模型,包含多个订单项
class Order(BaseModel):
    order_id: int
    customer_name: str
    # List[OrderItem] 表示 OrderItem 对象的列表
    items: List[OrderItem]

# 创建包含多个订单项的订单
order = Order(
    order_id=1001,
    customer_name="张三",
    items=[
        {"product_name": "笔记本电脑", "quantity": 1, "price": 5999.99},
        {"product_name": "鼠标", "quantity": 2, "price": 29.99},
        {"product_name": "键盘", "quantity": 1, "price": 99.99}
    ]
)
print(f"订单: {order}")
print(f"订单项数量: {len(order.items)}")
print(f"第一个订单项: {order.items[0].product_name}")

7. 高级功能 #

7.1 可选字段和默认值 #

我们已经在前面的例子中看到了可选字段,这里再详细说明。

创建文件 example_07_optional.py:

# ============================================
# 可选字段和默认值
# ============================================

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

# 定义用户模型,演示各种默认值用法
class UserProfile(BaseModel):
    # 必需字段:没有默认值,创建对象时必须提供
    username: str
    email: str

    # 可选字段1:使用 Optional 和 None 默认值
    # 如果不提供,值就是 None
    phone: Optional[str] = None

    # 可选字段2:有具体的默认值
    # 如果不提供,值就是 "普通用户"
    role: str = "普通用户"

    # 可选字段3:列表类型,默认是空列表
    tags: List[str] = []

    # 可选字段4:使用 Field 设置默认值和描述
    age: int = Field(18, ge=0, le=150, description="用户年龄")

    # 可选字段5:使用 Field 的 default_factory
    # default_factory 用于生成默认值(每次创建对象时调用)
    from datetime import datetime
    created_at: datetime = Field(default_factory=datetime.now)

# 示例1:只提供必需字段
print("=" * 50)
print("示例1:只提供必需字段")
print("=" * 50)
user1 = UserProfile(username="张三", email="zhangsan@example.com")
print(f"用户: {user1}")
print(f"电话(默认 None): {user1.phone}")
print(f"角色(默认值): {user1.role}")
print(f"标签(默认空列表): {user1.tags}")

# 示例2:提供部分可选字段
print("\n" + "=" * 50)
print("示例2:提供部分可选字段")
print("=" * 50)
user2 = UserProfile(
    username="李四",
    email="lisi@example.com",
    phone="13800138000",
    role="管理员"
)
print(f"用户: {user2}")

# 示例3:修改列表默认值(注意陷阱)
print("\n" + "=" * 50)
print("示例3:列表默认值的陷阱")
print("=" * 50)
# ⚠️ 注意:不要直接修改列表默认值!
# 错误示例(不要这样做):
# tags: List[str] = []  # 这会导致所有对象共享同一个列表

# 正确做法:使用 Field 的 default_factory
class CorrectUser(BaseModel):
    username: str
    # 使用 default_factory,每次创建对象时都会创建新列表
    tags: List[str] = Field(default_factory=list)

user3 = CorrectUser(username="王五")
user3.tags.append("VIP")  # 修改这个对象的 tags
print(f"用户3的标签: {user3.tags}")

user4 = CorrectUser(username="赵六")
print(f"用户4的标签(应该是空的): {user4.tags}")

7.2 数据转换和清洗 #

Pydantic 可以自动转换和清洗数据。

创建文件 example_08_transformation.py:

# ============================================
# 数据转换和清洗
# ============================================

from pydantic import BaseModel, validator
from typing import Optional

# 定义数据处理模型
class CleanedData(BaseModel):
    # 用户ID
    user_id: int
    # 用户名(会自动去除首尾空格)
    user_name: str
    # 邮箱(转换为小写)
    user_email: str

    # 验证器:自动清洗用户名
    @validator('user_name')
    def clean_name(cls, v):
        """
        清洗用户名:去除首尾空格,转换为标题格式
        """
        # strip() 去除首尾空格
        v = v.strip()
        # title() 将首字母大写
        v = v.title()
        return v

    # 验证器:自动清洗邮箱
    @validator('user_email')
    def clean_email(cls, v):
        """
        清洗邮箱:转换为小写,去除空格
        """
        # lower() 转换为小写
        v = v.lower()
        # replace() 去除空格
        v = v.replace(" ", "")
        return v

    # 配置:忽略额外字段
    class Config:
        # extra = "ignore" 表示忽略模型中没有定义的字段
        # 这样即使传入额外字段也不会报错
        extra = "ignore"

# 示例1:自动清洗数据
print("=" * 50)
print("示例1:自动清洗数据")
print("=" * 50)
# 输入数据有空格、大小写混乱
raw_data = {
    "user_id": 1,
    "user_name": "   zhang san   ",  # 有空格,小写
    "user_email": "  ZHANGSAN@EXAMPLE.COM  "  # 有空格,大写
}
cleaned = CleanedData(**raw_data)
print(f"原始数据: {raw_data}")
print(f"清洗后: {cleaned}")
print(f"用户名(已清洗): '{cleaned.user_name}'")
print(f"邮箱(已清洗): '{cleaned.user_email}'")

# 示例2:忽略额外字段
print("\n" + "=" * 50)
print("示例2:忽略额外字段")
print("=" * 50)
# 数据中包含模型中没有定义的字段
data_with_extra = {
    "user_id": 2,
    "user_name": "li si",
    "user_email": "lisi@example.com",
    "extra_field_1": "这个字段会被忽略",
    "extra_field_2": 12345
}
cleaned2 = CleanedData(**data_with_extra)
print(f"包含额外字段的数据: {data_with_extra}")
print(f"处理后的对象(额外字段已忽略): {cleaned2}")

# 示例3:实际应用场景
print("\n" + "=" * 50)
print("示例3:实际应用场景")
print("=" * 50)

def process_user_data(raw_data: dict):
    """
    处理用户数据的函数
    自动验证和清洗数据
    """
    try:
        # 自动验证和清洗
        cleaned_data = CleanedData(**raw_data)
        print(f"✓ 数据验证通过")
        print(f"  用户ID: {cleaned_data.user_id}")
        print(f"  用户名: {cleaned_data.user_name}")
        print(f"  邮箱: {cleaned_data.user_email}")
        return cleaned_data
    except Exception as e:
        print(f"✗ 数据验证失败: {e}")
        return None

# 测试处理函数
test_data = {
    "user_id": "3",  # 字符串,会自动转换为整数
    "user_name": "  wang wu  ",
    "user_email": "WANGWU@EXAMPLE.COM"
}
result = process_user_data(test_data)

8. 实际应用场景 #

8.1 API 数据验证(FastAPI 示例) #

FastAPI 是一个现代 Python Web 框架,它内置支持 Pydantic。

创建文件 example_09_fastapi.py:

# ============================================
# FastAPI 中使用 Pydantic(示例)
# ============================================
# 注意:运行此示例需要安装 fastapi 和 uvicorn
# Windows: pip install fastapi uvicorn
# Mac: pip3 install fastapi uvicorn

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr
from typing import Optional

# 创建 FastAPI 应用
app = FastAPI()

# 定义请求模型(客户端发送的数据)
class CreateUserRequest(BaseModel):
    # 用户名
    username: str
    # 密码
    password: str
    # 邮箱(使用 EmailStr 确保格式正确)
    email: EmailStr
    # 年龄(可选)
    age: Optional[int] = None

# 定义响应模型(服务器返回的数据)
class UserResponse(BaseModel):
    # 用户ID
    id: int
    # 用户名
    username: str
    # 邮箱
    email: str
    # 年龄
    age: Optional[int] = None

# 定义 API 端点
@app.post("/users/", response_model=UserResponse)
async def create_user(user: CreateUserRequest):
    """
    创建用户的 API 端点
    user 参数会自动使用 Pydantic 验证
    """
    # 这里 user 已经是验证过的 CreateUserRequest 对象
    # 可以直接使用,不需要手动验证

    # 模拟创建用户(实际应该保存到数据库)
    new_user = UserResponse(
        id=1,  # 实际应该从数据库获取
        username=user.username,
        email=user.email,
        age=user.age
    )

    # FastAPI 会自动将 UserResponse 转换为 JSON 返回
    return new_user

# 运行说明:
# 在终端运行: uvicorn example_09_fastapi:app --reload
# 然后访问: http://127.0.0.1:8000/docs 查看 API 文档
# 可以测试发送 POST 请求到 http://127.0.0.1:8000/users/

print("=" * 50)
print("FastAPI 示例代码")
print("=" * 50)
print("此文件定义了 FastAPI 应用和 Pydantic 模型")
print("要运行此示例,请执行以下步骤:")
print("1. 安装依赖: pip install fastapi uvicorn")
print("2. 运行服务器: uvicorn example_09_fastapi:app --reload")
print("3. 访问 http://127.0.0.1:8000/docs 查看 API 文档")

8.2 配置文件管理 #

使用 Pydantic 管理应用配置。

创建文件 example_10_config.py:

# ============================================
# 配置文件管理
# ============================================

from pydantic import BaseSettings
from typing import Optional

# 定义配置模型
# BaseSettings 是 Pydantic 提供的特殊基类,用于管理配置
class Settings(BaseSettings):
    # 应用名称,默认值是 "My App"
    app_name: str = "My App"
    # 数据库 URL(必需,没有默认值)
    database_url: str
    # 调试模式,默认是 False
    debug: bool = False
    # API 密钥(可选)
    api_key: Optional[str] = None
    # 最大连接数,默认是 10
    max_connections: int = 10

    # 配置类
    class Config:
        # 从 .env 文件加载环境变量
        env_file = ".env"
        # 环境变量文件编码
        env_file_encoding = "utf-8"

# 创建配置对象
# 会自动从环境变量和 .env 文件加载配置
try:
    settings = Settings()
    print("=" * 50)
    print("配置加载成功")
    print("=" * 50)
    print(f"应用名称: {settings.app_name}")
    print(f"数据库 URL: {settings.database_url}")
    print(f"调试模式: {settings.debug}")
    print(f"最大连接数: {settings.max_connections}")
except Exception as e:
    print("=" * 50)
    print("配置加载失败")
    print("=" * 50)
    print(f"错误: {e}")
    print("\n提示:")
    print("1. 创建 .env 文件,内容如下:")
    print("   database_url=postgresql://user:password@localhost/dbname")
    print("   app_name=My Application")
    print("   debug=true")
    print("2. 或者设置环境变量:")
    print("   Windows: set database_url=postgresql://...")
    print("   Mac: export database_url=postgresql://...")

# 示例:手动创建配置(不使用环境变量)
print("\n" + "=" * 50)
print("手动创建配置示例")
print("=" * 50)
manual_settings = Settings(
    app_name="手动配置的应用",
    database_url="sqlite:///example.db",
    debug=True,
    max_connections=20
)
print(f"手动配置: {manual_settings}")

创建 .env 文件(可选):

database_url=postgresql://user:password@localhost/mydb
app_name=My Application
debug=true
max_connections=20

8.3 数据验证和清洗工具 #

创建一个通用的数据验证工具。

创建文件 example_11_data_processor.py:

# ============================================
# 数据验证和清洗工具
# ============================================

from pydantic import BaseModel, validator, ValidationError
from typing import Optional, List, Dict, Any

# 定义用户数据模型
class UserData(BaseModel):
    # 用户ID
    user_id: int
    # 用户名
    user_name: str
    # 用户邮箱
    user_email: str
    # 创建时间(可选)
    created_at: Optional[str] = None

    # 配置:忽略额外字段
    class Config:
        extra = "ignore"

    # 验证用户名
    @validator('user_name')
    def validate_name(cls, v):
        v = v.strip()
        if not v:
            raise ValueError('用户名不能为空')
        if len(v) < 2:
            raise ValueError('用户名至少2个字符')
        return v

    # 验证邮箱
    @validator('user_email')
    def validate_email(cls, v):
        v = v.strip().lower()
        if '@' not in v:
            raise ValueError('邮箱格式不正确')
        return v

# 数据处理器类
class DataProcessor:
    """
    数据处理器,用于验证和清洗数据
    """

    @staticmethod
    def process_user_data(raw_data: dict) -> Dict[str, Any]:
        """
        处理用户数据

        参数:
        raw_data: 原始数据字典

        返回:
        处理后的数据字典,包含 'success', 'data', 'errors' 字段
        """
        try:
            # 使用 Pydantic 自动验证和清洗
            validated_data = UserData(**raw_data)

            # 转换为字典返回
            return {
                "success": True,
                "data": validated_data.dict(),
                "errors": None
            }
        except ValidationError as e:
            # 收集所有验证错误
            errors = []
            for error in e.errors():
                errors.append({
                    "field": ".".join(str(x) for x in error["loc"]),
                    "message": error["msg"],
                    "type": error["type"]
                })

            return {
                "success": False,
                "data": None,
                "errors": errors
            }
        except Exception as e:
            return {
                "success": False,
                "data": None,
                "errors": [{"message": str(e)}]
            }

# 示例使用
print("=" * 50)
print("数据处理器示例")
print("=" * 50)

# 测试数据1:正确的数据
print("\n测试1:正确的数据")
test_data_1 = {
    "user_id": 1,
    "user_name": "  张三  ",
    "user_email": "  ZHANGSAN@EXAMPLE.COM  "
}
result1 = DataProcessor.process_user_data(test_data_1)
print(f"成功: {result1['success']}")
if result1['success']:
    print(f"处理后的数据: {result1['data']}")
else:
    print(f"错误: {result1['errors']}")

# 测试数据2:错误的数据
print("\n测试2:错误的数据")
test_data_2 = {
    "user_id": "不是数字",
    "user_name": "A",  # 太短
    "user_email": "invalid-email"  # 格式错误
}
result2 = DataProcessor.process_user_data(test_data_2)
print(f"成功: {result2['success']}")
if not result2['success']:
    print("发现的错误:")
    for error in result2['errors']:
        print(f"  - {error['field']}: {error['message']}")

# 测试数据3:包含额外字段
print("\n测试3:包含额外字段(会被忽略)")
test_data_3 = {
    "user_id": 3,
    "user_name": "李四",
    "user_email": "lisi@example.com",
    "extra_field_1": "这个会被忽略",
    "extra_field_2": 12345
}
result3 = DataProcessor.process_user_data(test_data_3)
print(f"成功: {result3['success']}")
if result3['success']:
    print(f"处理后的数据(额外字段已忽略): {result3['data']}")

9. 常见问题 #

9.1 Q1: Pydantic 和普通 Python 类有什么区别? #

普通 Python 类:

class User:
    def __init__(self, name: str, age: int):
        self.name = name
        self.age = age

# 不会验证类型
user = User(name=123, age="不是数字")  # 不会报错,但类型错误

Pydantic 模型:

from pydantic import BaseModel

class User(BaseModel):
    name: str
    age: int

# 自动验证类型
user = User(name=123, age="不是数字")  # 会抛出 ValidationError

9.2 Q2: 如何查看模型的字段信息? #

from pydantic import BaseModel

class User(BaseModel):
    name: str
    age: int = 18

# 查看模型结构
print(User.schema())
# 或者
print(User.schema_json(indent=2))

9.3 Q3: 如何处理日期时间? #

from pydantic import BaseModel
from datetime import datetime

class Event(BaseModel):
    name: str
    # datetime 类型,Pydantic 会自动解析字符串
    created_at: datetime

# 可以使用字符串创建
event = Event(name="会议", created_at="2024-01-01 12:00:00")
print(event.created_at)  # 自动转换为 datetime 对象

9.4 Q4: 如何只验证部分字段? #

from pydantic import BaseModel, ValidationError

class User(BaseModel):
    name: str
    age: int
    email: str

# 使用 validate_assignment=False 可以部分更新
user = User(name="张三", age=25, email="zhangsan@example.com")
# 直接修改字段(默认会验证)
user.age = "不是数字"  # 会抛出错误

# 如果需要部分验证,可以使用 validate() 方法

9.5 Q5: Pydantic 性能如何? #

Pydantic 的性能很好,但对于大量数据的验证,可以考虑:

  • 使用 validate_assignment=False 关闭赋值验证
  • 缓存验证结果
  • 对于简单场景,可以考虑使用 dataclasses

10. 总结 #

  1. Pydantic 是什么

    • 基于类型注解的数据验证库
    • 自动验证、转换和清洗数据
  2. 主要功能

    • 类型验证:确保数据类型正确
    • 自动转换:尝试将数据转换为正确类型
    • 自定义验证:使用 @validator 定义验证规则
    • 序列化:轻松转换为字典或 JSON
  3. 适用场景

    • API 数据验证(如 FastAPI)
    • 配置文件管理
    • 数据清洗和转换
    • 任何需要验证数据的场景
  4. 优势

    • 代码简洁:定义一次,自动验证
    • 错误信息清晰:告诉你具体哪里错了
    • 易于维护:修改模型定义即可
    • 类型安全:基于 Python 类型提示

访问验证

请输入访问令牌

Token不正确,请重新输入