003.Pydantic

• 出处：Pydantic：目前最流行的Python数据验证库

• 原作者：暴走的海鸽

安装

安装 Pydantic 非常简单：

pip install pydantic[email]  # 会用到邮箱校验，直接在这一起安装了

如何使用 Pydantic？

使用 Pydantic 的主要方法是创建继承自 BaseModel 的自定义类，这是所有 Pydantic 模型的基类。然后，使用类型注释定义模型的属性，并选择性地提供默认值或验证器。

pydantic 的核心是模型（Model）

例如，让我们为用户创建一个简单的模型，并使用 Python 的类型注解来声明期望的数据类型：

#! -*-conding: UTF-8 -*-
from enum import Enum
from datetime import datetime
from typing import List, Optional
from pydantic import BaseModel, ValidationError, EmailStr
# 导入pydantic对应的模型基类
from pydantic import constr, conint

class GenderEnum(str, Enum):
    """
    性别枚举
    """
    male = "男"
    female = "女"

class User(BaseModel):
    id: int
    name: str = "小卤蛋"
    age: conint(ge=0, le=99)  # 整数范围：0 <= age <= 99
    email: EmailStr
    signup_ts: Optional[datetime] = None
    friends: List[str] = []
    password: constr(min_length=6, max_length=10)  # 字符长度
    phone: constr(pattern=r'^1\d{10}$')  # 正则验证手机号
    sex: GenderEnum  # 枚举验证, 能传: 男和女

验证数据

一旦定义了模型，可以使用它来验证数据。如果要从字典实例化 User 对象，可以使用 字典对象解包 或者 .model_validate()、.model_validate_json() 类方法：

if __name__ == '__main__':

    user_data = {
        "id": 123,
        "name": "小卤蛋",
        "age": 20,
        "email": "xiaoludan@example.com",
        'signup_ts': '2024-07-19 00:22',
        'friends': ["小明", '小天才', b''],
        'password': '123456',
        'phone': '13800000000',
        'sex': '男'
    }

    try:
        # user = User(**user_data)
        user = User.model_validate(user_data)
        print(f"User id: {user.id}, User name: {user.name}, User email: {user.email}")
    except ValidationError as e:
        print(f"Validation error: {e.json()}")

都符合模型定义的情况下，可以像往常一样访问模型的属性：

User id: 123, User name: 小卤蛋, User email: xiaoludan@example.com

如果数据不符合模型的定义(以下 故意不传 id 字段 )，Pydantic 将抛出一个 ValidationError。

Validation error: [{"type":"missing","loc":["id"],"msg":"Field required","input":{"name":"小卤蛋","age":20,"email":"xiaoludan@example.com","signup_ts":"2024-07-19 00:22","friends":["小明","小天才",""],"password":"123456","phone":"13800000000","sex":"男"},"url":"https://errors.pydantic.dev/2.8/v/missing"}]

自定义验证

除了内置的验证器，还可以为模型定义自定义验证器。假设要确保用户年龄在 18 岁以上，可以使用 @field_validator 装饰器创建一个自定义验证器：

# ! -*-conding: UTF-8 -*-
from datetime import datetime
from typing import List, Optional
from pydantic import BaseModel, EmailStr, field_validator, ValidationError

def check_name(v: str) -> str:
    """Validator to be used throughout"""
    if not v.startswith("小"):
        raise ValueError("must be startswith 小")
    return v

class User(BaseModel):
    id: int
    name: str = "小卤蛋"
    age: int
    email: EmailStr
    signup_ts: Optional[datetime] = None
    friends: List[str] = []

    validate_fields = field_validator("name")(check_name)

    @field_validator("age")
    @classmethod
    def check_age(cls, age):
        if age < 18:
            raise ValueError("用户年龄必须大于18岁")
        return age

当尝试创建一个只有12岁的小朋友用户：

Validation error: [{"type":"value_error","loc":["age"],"msg":"Value error, 用户年龄必须大于18岁","input":12,"ctx":{"error":"用户年龄必须大于18岁"},"url":"https://errors.pydantic.dev/2.8/v/value_error"}]

或者，当 name 不是 小 开头的话，将得到报错：

Validation error: [{"type":"value_error","loc":["name"],"msg":"Value error, must be startswith 小","input":"大卤蛋","ctx":{"error":"must be startswith 小"},"url":"https://errors.pydantic.dev/2.8/v/value_error"}]

如果要同时动态校验多个字段，还可以使用 model_validator 装饰器。

# ! -*-conding: UTF-8 -*-
# @公众号: 海哥python
from datetime import datetime
from typing import List, Optional, Self
from pydantic import BaseModel, ValidationError, EmailStr, field_validator, model_validator

def check_name(v: str) -> str:
    """Validator to be used throughout"""
    if not v.startswith("小"):
        raise ValueError("must be startswith 小")
    return v

class User(BaseModel):
    id: int
    name: str = "小卤蛋"
    age: int
    email: EmailStr
    signup_ts: Optional[datetime] = None
    friends: List[str] = []

    validate_fields = field_validator("name")(check_name)

    @field_validator("age")
    @classmethod
    def check_age(cls, age):
        if age < 18:
            raise ValueError("用户年龄必须大于18岁")
        return age

    @model_validator(mode="after")
    def check_age_and_name(self) -> Self:
        if self.age < 30 and self.name != "小卤蛋":
            raise ValueError("用户年龄必须小于30岁, 且名字必须为小卤蛋")

        return self

当尝试创建一个20岁的小小卤蛋用户：

# ! -*-conding: UTF-8 -*-
if __name__ == '__main__':
    user_data = {
        "id": 123,
        "name": "小小卤蛋",
        "age": 20,
        "email": "xiaoludan@example.com",
        'signup_ts': '2024-07-19 00:22',
        'friends': ["小明", '小天才', b''],
    }
    try:
        user = User(**user_data)
        print(user.model_dump())
    except ValidationError as e:
        print(f"Validation error: {e.json()}")

Validation error: [{"type":"value_error","loc":[],"msg":"Value error, 用户年龄必须小于30岁, 且名字必须为小卤蛋","input":{"id":123,"name":"小小卤蛋","age":20,"email":"xiaoludan@example.com","signup_ts":"2024-07-19 00:22","friends":["小明","小天才",""]},"ctx":{"error":"用户年龄必须小于30岁, 且名字必须为小卤蛋"},"url":"https://errors.pydantic.dev/2.8/v/value_error"}]

validate_call 也是非常有用的装饰器。

from typing import Annotated

from pydantic import BaseModel, Field, validate_call

class Person(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    age: int = Field(..., gt=0, lt=20)

@validate_call
def greet(person: Person, message: Annotated[str, Field(min_length=1, max_length=100)]):
    print(f"Hello, {person.name}! {message}")

# 错误的调用，将引发验证错误
try:
    greet(Person(name="小明", age=18), 1)
except Exception as e:
    print(e)

此时，执行会报错：

validation error for greet
  Input should be a valid string [type=string_type, input_value=1, input_type=int]
    For further information visit https://errors.pydantic.dev/2.5/v/string_type

计算属性

字段可能派生自其他字段，比如年龄一般会根据生日和当前日期动态计算得出、面积通过长和宽动态计算等。

#! -*-conding: UTF-8 -*-
from datetime import datetime
from typing import List, Optional
from pydantic import BaseModel, ValidationError, EmailStr, computed_field

class User(BaseModel):
    id: int
    name: str = "小卤蛋"
    age: int
    email: EmailStr
    signup_ts: Optional[datetime] = None
    friends: List[str] = []

    @computed_field  # 计算属性
    @property
    def link(self) -> str:
        return f"尼古拉斯 · {self.name}"

if __name__ == '__main__':

    user_data = {
        "id": 123,
        "name": "小卤蛋",
        "age": 20,
        "email": "xiaoludan@example.com",
        'signup_ts': '2024-07-19 00:22',
        'friends': ["小明", '小天才', b''],
    }
    #
    try:
        user = User(**user_data)
        print(f"{user.model_dump()} .... type: {type(user.model_dump())}")
    except ValidationError as e:
        print(f"Validation error: {e.json()}")

输出结果为（序列化后会发现多了 link 字段）：

{'id': 123, 'name': '小卤蛋', 'age': 20, 'email': 'xiaoludan@example.com', 'signup_ts': datetime.datetime(2024, 7, 19, 0, 22), 'friends': ['小明', '小天才', ''], 'link': '尼古拉斯 · 小卤蛋'} .... type: <class 'dict'>

管理配置

pip install pydantic_settings 

使用 Pydantic 的 BaseSettings 可以很方便的管理应用程序的配置。

# ! -*-conding: UTF-8 -*-
import os

# 从pydantic模块导入HttpUrl和Field类，用于设置和验证配置数据的类型和约束
from pydantic import HttpUrl, Field
# 从pydantic_settings模块导入BaseSettings类，作为配置类的基类
from pydantic_settings import BaseSettings

# 初始化环境变量，这些环境变量将用于配置应用程序的数据库和API访问
os.environ['DATABASE_HOST'] = "http://baidu.com"
os.environ['DATABASE_USER'] = "小明"
os.environ['DATABASE_PASSWORD'] = "123456abcd"
os.environ['API_KEY'] = "DHKSDsdh*(sdds"

class AppConfig(BaseSettings):
    """
    应用程序配置类，继承自BaseSettings，用于管理应用程序的配置信息。
    Attributes:
        database_host: 数据库主机的URL，必须是一个有效的HTTP或HTTPS URL。
        database_user: 数据库用户的名称，最小长度为5个字符。
        database_password: 数据库用户的密码，最小长度为10个字符。
        api_key: API访问的密钥，最小长度为8个字符。
    """
    # 定义配置项database_host，类型为HttpUrl，确保其为有效的HTTP或HTTPS URL
    database_host: HttpUrl
    # 定义配置项database_user，类型为字符串，默认最小长度为5
    database_user: str = Field(min_length=5)
    # 定义配置项database_password，类型为字符串，默认最小长度为10
    database_password: str = Field(min_length=10)
    # 定义配置项api_key，类型为字符串，默认最小长度为8
    api_key: str = Field(min_length=8)

# 打印配置类的实例化对象的模型信息，用于调试和确认配置的正确性
print(AppConfig().model_dump())

执行结果：

{'database_host': Url('http://baidu.com/'), 'database_user': '小明', 'database_password': '123456abcd', 'api_key': 'DHKSDsdh*(sdds'}

如果是配置文件等，则可以通过 model_config 进行配置。

新建一个 .env 配置文件

DATABASE_HOST=http://baidu.com
DATABASE_USER=小明
DATABASE_PASSWORD=123456abcd
API_KEY=DHKSDsdh*(sdds

# ! -*-conding: UTF-8 -*-
# 导入Pydantic的HttpUrl和Field类，用于配置验证
from pydantic import HttpUrl, Field
# 导入BaseSettings和SettingsConfigDict类，用于设置配置类的基础行为和配置字典
from pydantic_settings import BaseSettings, SettingsConfigDict

class AppConfig(BaseSettings):
    """
    应用配置类，继承自BaseSettings，用于定义和管理应用的配置项。
    Attributes:
        model_config: 配置模型的设置，用于指定.env文件的位置、编码方式、是否大小写敏感以及额外的配置策略。
        database_host: 数据库主机的URL，必须是一个有效的HTTP或HTTPS URL。
        database_user: 数据库用户的名称，最小长度为5个字符。
        database_password: 数据库用户的密码，最小长度为10个字符。
        api_key: API的密钥，最小长度为8个字符。
    """
    # 定义配置模型的设置，包括.env文件位置、编码、大小写敏感性和额外参数策略
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
        extra="forbid",
    )

    # 数据库主机的URL，必须是一个有效的HTTP或HTTPS URL
    database_host: HttpUrl
    # 数据库用户的名称，最小长度为5个字符
    database_user: str = Field(min_length=5)
    # 数据库用户的密码，最小长度为10个字符
    database_password: str = Field(min_length=10)
    # API的密钥，最小长度为8个字符
    api_key: str = Field(min_length=8)

# 打印配置类的实例化对象的模型信息，用于调试和确认配置的正确性
print(AppConfig().model_dump())

嵌套数据模型

Pydantic 支持嵌套的数据模型，方便管理复杂的数据结构。

#! -*-conding: UTF-8 -*-
from typing import List
from pydantic import BaseModel, conint

class Friend(BaseModel):
    name: str
    age: conint(gt=0, le=99)

class User(BaseModel):
    name: str
    age: conint(gt=0, le=99)
    friends: List[Friend]

# 创建并验证数据
user_data = {
    'name': '小明',
    'age': 30,
    'friends': [{'name': '小卤蛋', 'age': 3}, {'name': '李元芳', 'age': 18}]
}
user = User(**user_data)
print(user)  # name='小明' age=30 friends=[Friend(name='小卤蛋', age=3), Friend(name='李元芳', age=18)]

Field 对象

Pydantic 的 Field 函数是一个强大的工具，它允许你在模型字段上设置额外的验证规则和默认值。Field 函数通常与模型字段一起使用，以提供更多的定制选项。

以下是一些常用的参数：

• 推荐使用 default_factory 为 list , dict , datetime 等可变类型提供默认值
• 避免 default=[] 这种共享引用问题

#! -*-conding: UTF-8 -*-
from pydantic import BaseModel, Field, EmailStr, ValidationError, SecretStr
from typing import List, Optional
from datetime import datetime

class User(BaseModel):
    id: int = Field(..., alias="_id", frozen=True, strict=True)  # 设置别名，创建后id不能被修改，id不能是字符串形式的“123”传入
    name: str = Field(default="小卤蛋", min_length=1, max_length=100)  # 设置默认值，使用 min_length 和 max_length 来限制字符串长度
    age: int = Field(gt=0)  # 支持各类条件验证，这里假设年龄必须大于0
    email: EmailStr
    signup_ts: Optional[datetime] = Field(default_factory=datetime.now, nullable=False, validate_default=True)
    friends: List[str] = Field(default=[], min_items=0)
    passwd: SecretStr = Field(min_length=6, max_length=20, exclude=True)  # passwd不会被序列化

if __name__ == '__main__':
    print(User.model_json_schema())

    user_data = {
        "_id": 123,  # 使用别名 _id
        "name": "小卤蛋",
        "age": 20,
        "email": "xiaoludan@example.com",
        # 'signup_ts': '2024-07-19 00:22',
        'friends': ["小明", '小天才', b''],
        "passwd": "123456"
    }

    try:
        user = User(**user_data)
        print(f"创建用户: {user}") # 
        print(f"转成字典形式： {user.model_dump()} .... type: {type(user.model_dump())}")
        print(f"转成json格式：{user.model_dump_json()} .... type: {type(user.model_dump_json())}")
        print(f"用户属性： User id: {user.id}, User name: {user.name}, User email: {user.email}")
        # user.id = 456   # 这里修改会报错
    except ValidationError as e:
        print(f"Validation error: {e.json()}")

将得到结果：

{'properties': {'_id': {'title': ' Id', 'type': 'integer'}, 'name': {'default': '小卤蛋', 'maxLength': 100, 'minLength': 1, 'title': 'Name', 'type': 'string'}, 'age': {'exclusiveMinimum': 0, 'title': 'Age', 'type': 'integer'}, 'email': {'format': 'email', 'title': 'Email', 'type': 'string'}, 'signup_ts': {'anyOf': [{'format': 'date-time', 'type': 'string'}, {'type': 'null'}], 'nullable': False, 'title': 'Signup Ts'}, 'friends': {'default': [], 'items': {'type': 'string'}, 'minItems': 0, 'title': 'Friends', 'type': 'array'}, 'passwd': {'maxLength': 20, 'minLength': 6, 'title': 'Passwd', 'type': 'string'}}, 'required': ['_id', 'age', 'email', 'passwd'], 'title': 'User', 'type': 'object'}
创建用户: id=123 name='小卤蛋' age=20 email='xiaoludan@example.com' signup_ts=datetime.datetime(2024, 7, 23, 11, 22, 46, 137194) friends=['小明', '小天才', ''] passwd='123456'
转成字典形式： {'id': 123, 'name': '小卤蛋', 'age': 20, 'email': 'xiaoludan@example.com', 'signup_ts': datetime.datetime(2024, 7, 23, 11, 22, 46, 137194), 'friends': ['小明', '小天才', '']} .... type: <class 'dict'>
转成json格式：{"id":123,"name":"小卤蛋","age":20,"email":"xiaoludan@example.com","signup_ts":"2024-07-23T11:22:46.137194","friends":["小明","小天才",""]} .... type: <class 'str'>
用户属性： User id: 123, User name: 小卤蛋, User email: xiaoludan@example.com

Config 配置选项

如果要对 BaseModel 中的某一基本型进行统一的格式要求，我们还可以使用 Config 类来实现。

以下是一些 Config 类中常见的属性及其含义：

#! -*-conding: UTF-8 -*-
from pydantic import BaseModel

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

    class Config:
        str_min_length = 10  # 字符串最小长度
        str_max_length = 20  # 字符串最大长度

user = User(name="John Doe", age=30)

执行将得到结果：

Validation error: [{"type":"string_too_short","loc":["name"],"msg":"String should have at least 10 characters","input":"John Doe","ctx":{"min_length":10},"url":"https://errors.pydantic.dev/2.5/v/string_too_short"}]

序列化

使用 模型类.model_dump() 方法可以将一个模型类实例对象转换为字典类型数据。

#! -*-conding: UTF-8 -*-
from datetime import datetime
from typing import List, Optional
from pydantic import BaseModel, ValidationError, EmailStr, field_validator, field_serializer
from enum import Enum

class GenderEnum(str, Enum):
    """
    性别枚举
    """
    male = "男"
    female = "女"

class User(BaseModel):
    id: int
    name: str = "小卤蛋"
    age: int
    email: EmailStr
    signup_ts: Optional[datetime] = datetime.now()
    friends: List[str] = []
    sex: GenderEnum

    @field_validator("age")
    @classmethod
    def check_age(cls, age):
        if age < 18:
            raise ValueError("用户年龄必须大于18岁")
        return age

    @field_serializer('signup_ts', when_used="always")
    def serialize_signup_ts(self, value: datetime) -> str:
        return value.strftime('%Y-%m-%d %H:%M:%S')

    @field_serializer('sex', when_used="always")
    def serialize_sex(self, value) -> str:
        return value.value

if __name__ == '__main__':
    user_data = {
        "id": 123,
        "name": "小卤蛋",
        "age": 20,
        "email": "xiaoludan@example.com",
        # 'signup_ts': '2024-07-19 00:22',
        'friends': ["公众号：海哥python", '小天才', b''],
        "sex": "男",
    }

    try:
        user = User.model_validate(user_data)
        print(f"{user.model_dump()} .... type: {type(user.model_dump())}")
    except ValidationError as e:
        print(f"Validation error: {e.json()}")

默认情况下， datetime 对象被序列化为 ISO 8601 字符串。这里使用 field_serializer 自定义序列化规则。

{'id': 123, 'name': '小卤蛋', 'age': 20, 'email': 'xiaoludan@example.com', 'signup_ts': '2024-07-24 14:47:33', 'friends': ['公众号：海哥python', '小天才', ''], 'sex': '男'} .... type: <class 'dict'>

使用模型类 .model_dump_json() 方法可以将一个模型类实例对象转换为 JSON 字符串。

生成文档

Pydantic 可以自动生成 API 文档

#! -*-conding: UTF-8 -*-
from datetime import datetime
from typing import List, Optional
from pydantic import BaseModel, EmailStr, field_validator

class User(BaseModel):
    id: int
    name: str = "小卤蛋"
    age: int
    email: EmailStr
    signup_ts: Optional[datetime] = None
    friends: List[str] = []

    @field_validator("age")
    def check_age(cls, age):
        if age < 18:
            raise ValueError("用户年龄必须大于18岁")
        return age

if __name__ == '__main__':
    print(User.model_json_schema())

通过 model_json_schema 方法可以得到 API 文档。

{'properties': {'id': {'title': 'Id', 'type': 'integer'}, 'name': {'default': '小卤蛋', 'title': 'Name', 'type': 'string'}, 'age': {'title': 'Age', 'type': 'integer'}, 'email': {'format': 'email', 'title': 'Email', 'type': 'string'}, 'signup_ts': {'anyOf': [{'format': 'date-time', 'type': 'string'}, {'type': 'null'}], 'default': None, 'title': 'Signup Ts'}, 'friends': {'default': [], 'items': {'type': 'string'}, 'title': 'Friends', 'type': 'array'}}, 'required': ['id', 'age', 'email'], 'title': 'User', 'type': 'object'}

应用场景

LangChain 中使用 Pydantic

以下，我们借助 LangChain 和 Pydantic 实现一个基于通义千问模型的问答路由系统。

新建 .env 文件：

DASHSCOPE_API_KEY=sk-b920635xxxdsdsdsd7edc2590sds20300  # 换成自己的

根据用户的问题类型，选择向量存储或网络搜索来提供答案，并将答案解析为结构化的数据：

# 导入环境变量加载工具和必要的库
import os
from dotenv import find_dotenv, load_dotenv
from typing import Literal

# 导入通义千问模型相关的类和模块
from langchain_community.chat_models import QianfanChatEndpoint
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI

# 加载环境变量
load_dotenv(find_dotenv())
# 从环境变量中获取DashScope的API密钥
DASHSCOPE_API_KEY = os.environ["DASHSCOPE_API_KEY"]

# 定义一个模型，用于路由用户查询到合适的数据库
# 定义 数据格式
class RouteQuery(BaseModel):
    """用于路由用户查询的模型，决定查询应路由到vectorstore还是web_search。

    Attributes:
        datasource (Literal["vectorstore", "web_search"]): 查询应路由到的数据源，可以是"vectorstore"或"web_search"。
    """
    datasource: Literal["vectorstore", "web_search"] = Field(
        ...,
        description="根据用户问题选择将其路由到网络搜索还是vectorstore。",
    )

if __name__ == '__main__':
    # 初始化通义千问模型
    api_key = DASHSCOPE_API_KEY
    qwen_model = ChatOpenAI(
        model_name="qwen-max",
        openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1",
        openai_api_key=api_key,
    )
    llm = qwen_model

    # 初始化输出解析器，用于解析模型的输出
    parser = PydanticOutputParser(pydantic_object=RouteQuery)

    # 获取数据格式说明
    data_pattern = parser.get_format_instructions()

    # 定义路由提示模板
    # Prompt
    system = """你是一个用户查询路由专家，负责将用户查询路由到vectorstore或web搜索。
    vectorstore包含有关代理、prompt工程和对抗攻击的文档。
    对于这些主题的问题，使用vectorstore。否则，使用web搜索。
    生成格式化数据模式如下：
    {data_pattern}  
    """
    route_prompt = ChatPromptTemplate.from_messages(
        [
            ("system", system),
            ("human", "{question}"),
        ]
    )

    # 构建查询路由管道
    question_router = route_prompt | llm

    # 调用查询路由管道并获取结果
    # 示例问题：小明是谁？
    res = question_router.invoke({"question": "小明是谁？", "data_pattern": data_pattern})
    # res = question_router.invoke({"question": "如何向代理添加记忆?", "data_pattern": data_pattern})

    # 解析获取的结果
    content = res.content
    print(content)

与 FastAPI 的深度集成

在 FastAPI 中，Pydantic 模型可自动处理请求解析与响应序列化：

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float
    description: str | None = None

@app.post("/items/")
async def create_item(item: Item):
    return {
        "message": "Item received",
        "item": item.model_dump()
    }

无需手动处理任何参数校验逻辑，FastAPI 会自动根据 Pydantic 模型生成 OpenAPI 文档和数据验证规则。

调试与错误处理

from pydantic import ValidationError

try:
    User(email="invalid")
except ValidationError as e:
    print(e.json(indent=2))

Pydantic 的错误对象支持 .errors() , .json() , .display() 等方法，便于日志、响应和开发调试。