Pydantic:目前最流行的Python数据验证库

CSDN 2024-10-24 11:35:02 阅读 63

前言

在处理来自系统外部的数据,如API、终端用户输入或其他来源时,我们必须牢记开发中的一条基本原则:“永远不要相信用户的输入”。

因此,我们必须对这些数据进行严格的检查和验证,确保它们被适当地格式化和标准化。这样做的目的是为了确保这些数据符合我们的程序所需的输入规范,从而保障项目能够正确且高效地运行。

为什么使用 Python 的 Pydantic 库?

Pydantic 是一个在 Python 中用于数据验证和解析的第三方库,<code>它现在是 Python 使用最广泛的数据验证库。

它利用声明式的方式定义数据模型Python 类型提示的强大功能来执行数据验证和序列化,使您的代码更可靠、更可读、更简洁且更易于调试。。它还可以从模型生成 JSON 架构,提供了自动生成文档等功能,从而轻松与其他工具集成。

Pydantic 在很多优秀的项目中被广泛使用。

Pydantic 的一些主要特性

易用性

Pydantic 使用起来简单直观,需要最少的样板代码和配置。它适用于许多流行的 IDE 和静态分析工具,例如 PyCharm、VS Code、mypy 等。Pydantic 可以轻松与其他流行的 Python 库(如 Flask、Django、FastAPI 和 SQLAlchemy)集成,使其易于在现有项目中使用。


类型注解

Pydantic 使用类型注解来定义模型的字段类型,以确保确保数据符合预期的类型和格式。你可以使用<code>Python 内置的类型、自定义类型或者其他Pydantic 提供的验证类型


数据验证,用户友好的错误

Pydantic 自动根据模型定义进行数据验证。它会检查字段的类型、长度、范围等,并自动报告验证错误,Pydantic 会提供信息丰富且可读的错误消息,包括错误的位置、类型和输入。你可以使用 ValidationError 异常来捕获验证错误。


序列化与反序列化

Pydantic 提供了从各种数据格式(例如 JSON、字典)到模型实例的转换功能。它可以自动将输入数据解析成模型实例,并保留类型安全性和验证规则。


性能高

Pydantic 的核心验证逻辑是用 Rust 编写的,使其成为 Python 中最快的数据验证库之一。它还支持延迟验证和缓存,以提高效率。

Pydantic和内置的dataclasses非常的像,主要区别在于Pydantic拥有更加强大的数据验证和序列化功能。


安装

安装 Pydantic 非常简单:

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

如何使用 Pydantic?

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

pydantic的核心是模型(Model)

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

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

# @公众号: 海哥python

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的类,继承自BaseModel。

id属性是整型,且是必需的,表示用户ID。name属性是字符串类型,默认值为’小卤蛋’。age属性是整型,且是必需的,表示用户年龄。email属性是电子邮件地址类型。signup_ts属性是可选的日期时间类型,默认值为None,表示用户注册时间。friends属性是字符串列表类型,默认值为空列表,表示用户的朋友列表。sex属性是枚举类型,可选值为“男”或“女”,表示用户的性别。

验证数据

一旦你定义了模型,你可以使用它来验证数据。

如果要从字典实例化 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': ["公众号:海哥python", '小天才', 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。

if __name__ == '__main__':

user_data = {

# "id": 123,

"name": "小卤蛋",

"age": 20,

"email": "xiaoludan@example.com",

'signup_ts': '2024-07-19 00:22',

'friends': ["公众号:海哥python", '小天才', 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()}")

报错:

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":["公众号:海哥python","小天才",""],"password":"123456","phone":"13800000000","sex":"男"},"url":"https://errors.pydantic.dev/2.8/v/missing"}]

自定义验证

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

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

# @公众号: 海哥python

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岁的小朋友用户:

if __name__ == '__main__':

user_data = {

"id": 123,

"name": "小卤蛋",

"age": 12,

"email": "xiaoludan@example.com",

'signup_ts': '2024-07-19 00:22',

'friends': ["公众号:海哥python", '小天才', b''],

}

try:

user = User(**user_data)

except ValidationError as e:

print(f"Validation error: { e.json()}")

将得到一个错误:

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不是开头的话:

if __name__ == '__main__':

user_data = {

"id": 123,

"name": "大卤蛋",

"age": 20,

"email": "xiaoludan@example.com",

'signup_ts': '2024-07-19 00:22',

'friends': ["公众号:海哥python", '小天才', b''],

}

try:

user = User(**user_data)

except ValidationError as e:

print(f"Validation error: { e.json()}")

将得到报错:

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

from typing_extensions import Self # 如果python版本不低于3.11,则可以直接从typing中导入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")code>

def check_age_and_name(self) -> Self:

if self.age < 30 and self.name != "小卤蛋":

raise ValueError("用户年龄必须小于30岁, 且名字必须为小卤蛋")

return self

if __name__ == '__main__':

user_data = { -- -->

"id": 123,

"name": "小小卤蛋",

"age": 20,

"email": "xiaoludan@example.com",

'signup_ts': '2024-07-19 00:22',

'friends': ["公众号:海哥python", '小天才', 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":["公众号:海哥python","小天才",""]},"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}")

# 正确的调用

greet(Person(name="公众号:海哥python", age=18), "How are you?")code>

greet(Person(name="公众号:海哥python", age=18), 1)code>

在不使用<code>validate_call的情况下,虽然pycharm中会提示1类型不匹配,但是实际执行时并不会报错。

然而,我们通常是希望定义和使用要符合我们的预期,以避免不可预见的错误。

此时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="公众号:海哥python", age=18), 1)code>

except Exception as e:

print(e)

此时,执行会报错:

1 validation error for greet

1

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

计算属性

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

以下我们动态增加link字段为例:

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

# @公众号: 海哥python

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': ["公众号:海哥python", '小天才', 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': ['公众号:海哥python', '小天才', ''], 'link': '尼古拉斯 · 小卤蛋'} .... type: <class 'dict'>

管理配置

pip install pydantic_settings

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

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

# @公众号: 海哥python

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'] = "公众号:海哥python"

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': '公众号:海哥python', 'database_password': '123456abcd', 'api_key': 'DHKSDsdh*(sdds'}


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

新建一个.env配置文件:

DATABASE_HOST=http://baidu.com

DATABASE_USER=公众号:海哥python

DATABASE_PASSWORD=123456abcd

API_KEY=DHKSDsdh*(sdds

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

# @公众号: 海哥python

# 导入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",code>

env_file_encoding="utf-8",code>

case_sensitive=False,

extra="forbid",code>

)

# 数据库主机的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())

执行结果:

{ -- -->'database_host': Url('http://baidu.com/'), 'database_user': '公众号:海哥python', 'database_password': '123456abcd', 'api_key': 'DHKSDsdh*(sdds'}

嵌套数据模型

Pydantic 支持嵌套的数据模型,方便管理复杂的数据结构。以下是一个示例代码:

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

# @公众号: 海哥python

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': '公众号:海哥python',

'age': 30,

'friends': [{ 'name': '小卤蛋', 'age': 3}, { 'name': '李元芳', 'age': 18}]

}

user = User(**user_data)

print(user) # name='公众号:海哥python' age=30 friends=[Friend(name='小卤蛋', age=3), Friend(name='李元芳', age=18)]code>

Field 对象

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

以下是一些常用的参数:

参数 具体含义
... 表示该字段是必填项
default 用于定义字段的默认值
default_factory 用于定义字段的默认值函数
alias 字段定义别名
validation_alias 字段定义别名,只想将别名用于验证
serialization_alias 字段定义别名,只想定义用于序列化的别名
gtltge 约束数值,大于、小于、大于或等于
min_lengthmax_length 约束字符串
min_itemsmax_items 元组、列表或集合约束
validate_default 控制是否应验证字段的默认值,默认情况下,不验证字段的默认值。
strict 指定是否应在“严格模式”下验证字段
frozen 用于模拟冻结的数据类行为
exclude 用于控制导出模型时应从模型中排除哪些字段
pattern 对于字符串字段,您可以设置为 pattern 正则表达式以匹配该字段所需的任何模式。

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

# @公众号: 海哥python

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”传入code>

name: str = Field(default="小卤蛋", min_length=1, max_length=100) # 设置默认值,使用 min_length 和 max_length 来限制字符串长度code>

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': ["公众号:海哥python", '小天才', 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=['公众号:海哥python', '小天才', ''] passwd='123456'code>

转成字典形式: { -- -->'id': 123, 'name': '小卤蛋', 'age': 20, 'email': 'xiaoludan@example.com', 'signup_ts': datetime.datetime(2024, 7, 23, 11, 22, 46, 137194), 'friends': ['公众号:海哥python', '小天才', '']} .... type: <class 'dict'>

转成json格式:{ "id":123,"name":"小卤蛋","age":20,"email":"xiaoludan@example.com","signup_ts":"2024-07-23T11:22:46.137194","friends":["公众号:海哥python","小天才",""]} .... type: <class 'str'>

用户属性: User id: 123, User name: 小卤蛋, User email: xiaoludan@example.com

Config 配置选项

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

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

参数 取值类型 具体含义
str_min_length int str 类型的最小长度,默认值为None
str_max_length int str 类型的最大长度。默认值为None
extra str 在模型初始化期间是否忽略、允许或禁止额外的属性。默认值为 ‘ignore’。allow - 允许任何额外的属性。forbid - 禁止任何额外的属性。ignore - 忽略任何额外的属性。
frozen bool 模型是否可变
str_to_upper bool 是否将 str 类型的所有字符转换为大写。默认值为 False 。
str_strip_whitespace bool 是否去除 str 类型的前导和尾随空格。
str_to_lower bool 是否将 str 类型的所有字符转换为小写。默认值为 False 。

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

# @公众号: 海哥python

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)code>

执行将得到结果:

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"}]

有关Config类中的特殊关键词名称,这里只给出了两个简单的例子,更多的内容可以参考官网中的文档说明。

序列化

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

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

# @公众号: 海哥python

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")code>

def serialize_signup_ts(self, value: datetime) -> str:

return value.strftime('%Y-%m-%d %H:%M:%S')

@field_serializer('sex', when_used="always")code>

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字符串。

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

# @公众号: 海哥python

from datetime import datetime

from typing import List, Optional, Any

from pydantic import BaseModel, ValidationError, EmailStr, field_validator, field_serializer, model_serializer

class User(BaseModel):

id: int

name: str = "小卤蛋"

age: int

email: EmailStr

signup_ts: Optional[datetime] = datetime.now()

friends: List[str] = []

@field_validator("age")

@classmethod

def check_age(cls, age):

if age < 18:

raise ValueError("用户年龄必须大于18岁")

return age

@field_serializer('signup_ts', when_used="json")code>

def serialize_signup_ts(self, value: datetime) -> str:

return value.strftime('%Y-%m-%d %H:%M:%S')

@model_serializer(when_used="json")code>

def serialize_model(self) -> dict[str, Any]:

return { -- -->

'id': self.id,

'name': self.name,

'age': self.age + 1,

'email': self.email,

'signup_ts': self.serialize_signup_ts(self.signup_ts),

'friends': self.friends,

}

if __name__ == '__main__':

user_data = {

"id": 123,

"name": "小卤蛋",

"age": 20,

"email": "xiaoludan@example.com",

# 'signup_ts': '2024-07-19 00:22',

'friends': ["公众号:海哥python", '小天才', b''],

}

try:

user = User.model_validate(user_data)

print(f"{ user.model_dump_json()} .... type: { type(user.model_dump_json())}")

print(f"{ user.model_dump()} .... type: { type(user.model_dump())}")

except ValidationError as e:

print(f"Validation error: { e.json()}")

也可以使用model_serializer对整体模型的序列化做定制。结果如下:

{ "id":123,"name":"小卤蛋","age":21,"email":"xiaoludan@example.com","signup_ts":"2024-07-24 14:17:42","friends":["公众号:海哥python","小天才",""]} .... type: <class 'str'>

{ 'id': 123, 'name': '小卤蛋', 'age': 20, 'email': 'xiaoludan@example.com', 'signup_ts': datetime.datetime(2024, 7, 24, 14, 17, 42, 45474), 'friends': ['公众号:海哥python', '小天才', '']} .... type: <class 'dict'>

生成文档

Pydantic 可以自动生成 API 文档。

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

# @公众号: 海哥python

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'}

应用场景

Flask集成Pydantic

以 Flask 为例,我们亦可以很方便的集成Pydantic:

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

# @公众号: 海哥python

from functools import wraps

from flask import Flask, request, jsonify

from pydantic import BaseModel, ValidationError

app = Flask(__name__)

# 创建一个 Pydantic 模型来表示请求体中的数据

class User(BaseModel):

model_config = { "extra": "forbid"} # 不允许额外字段

id: int

name: str

email: str

def validate_request_body(model):

def decorator(func):

@wraps(func)

def wrapper(*args, **kwargs):

try:

body = model.model_validate(request.json)

return func(body, *args, **kwargs)

except ValidationError as e:

return jsonify({ "error": e.errors()}), 400

return wrapper

return decorator

@app.route('/users', methods=['POST'])

@validate_request_body(User)

def create_user2(user: User):

# 在这里可以处理用户数据,例如保存到数据库

# ...

print(user.model_dump())

# 返回成功响应

return jsonify({ "message": "User created successfully", "user": user.dict()}), 201

if __name__ == '__main__':

app.run(debug=True)

参数验证失败时:

curl -X POST -H "Content-Type: application/json" -d '{"id": 1, "name": 1, "email": "john.doe@example.com"}' http://localhost:5000/users

{

"error": [

{

"input": 1,

"loc": [

"name"

],

"msg": "Input should be a valid string",

"type": "string_type",

"url": "https://errors.pydantic.dev/2.8/v/string_type"

}

]

}

所有验证错误,包括 ValueError 等我们在自定义验证器中提出的错误,都包含在 Pydantic 的 ValidationError。因此,一种常见的做法是为其设置全局错误处理程序。

flask_pydantic也为您提供一种集成FalskPydantic的现成方案,感兴趣的话可以自行研究。


LangChain中使用Pydantic

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

新建.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。",code>

)

if __name__ == '__main__':

# 初始化通义千问模型

api_key = DASHSCOPE_API_KEY

qwen_model = ChatOpenAI(

model_name="qwen-max",code>

openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1",code>

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

# 调用查询路由管道并获取结果

# 示例问题:公众号:海哥python 有多少粉丝?

res = question_router.invoke({ -- -->"question": "公众号:海哥python 有多少粉丝?", "data_pattern": data_pattern})

# res = question_router.invoke({"question": "如何向代理添加记忆?", "data_pattern": data_pattern})

# 解析获取的结果

content = res.content

print(content)

这样,我们就得到一个结构化的数据:

{ "datasource": "web_search"}

后续便可以根据这个统一的结构化数据选择适合的源进行检索。

小结

目前,Python 中主流的数据验证库还有不少,但是 Pydantic 无疑是那个星星增长最迅猛的。

总而言之,Pydantic 是一个灵活而强大的数据验证库,它利用 Python 类型注解来简化数据验证和解析。无论是处理用户输入、验证数据库查询结果,还是解析 API 响应,都能提供直观而高效的解决方案。

Pydantic 的强大远不止这些,更多使用技巧请查阅官方文档 …

最后

你是否也在项目中使用过<code>Pydantic吗?您以什么形式使用呢?对于Pyddantic您有更多的心得吗?欢迎在评论区告诉我。

今天的分享就到这里。如果觉得不错,点赞在看关注安排起来吧。

参考

https://mp.weixin.qq.com/s/uB8ff3sJjz2fx1n5cmckaw

https://mp.weixin.qq.com/s/wAUxaZOjPIaaeLWCH9Tk7g

https://realpython.com/python-pydantic/

https://shzhangji.com/blog/2024/01/28/define-api-data-models-with-pydantic/

https://www.netguru.com/blog/data-validation-pydantic

https://jsontopydantic.com/



声明

本文内容仅代表作者观点,或转载于其他网站,本站不以此文作为商业用途
如有涉及侵权,请联系本站进行删除
转载本站原创文章,请注明来源及作者。