Python使用Pydantic進(jìn)行數(shù)據(jù)驗(yàn)證與序列化詳解
1. 引言
在當(dāng)今的數(shù)據(jù)驅(qū)動(dòng)世界中,確保數(shù)據(jù)的一致性和完整性是軟件開(kāi)發(fā)中的關(guān)鍵挑戰(zhàn)。Python作為一門動(dòng)態(tài)類型語(yǔ)言,雖然靈活性強(qiáng),但在處理復(fù)雜數(shù)據(jù)結(jié)構(gòu)和API交互時(shí),常常面臨類型錯(cuò)誤和數(shù)據(jù)驗(yàn)證的問(wèn)題。Pydantic庫(kù)應(yīng)運(yùn)而生,它通過(guò)使用Python類型注解來(lái)提供數(shù)據(jù)驗(yàn)證和設(shè)置管理,使得數(shù)據(jù)處理變得更加可靠和高效。
Pydantic的核心優(yōu)勢(shì)在于:
- 運(yùn)行時(shí)類型檢查:在數(shù)據(jù)解析和實(shí)例化時(shí)進(jìn)行類型驗(yàn)證
- 數(shù)據(jù)序列化:輕松將Python對(duì)象轉(zhuǎn)換為JSON、字典等格式
- 配置管理:統(tǒng)一的數(shù)據(jù)配置和驗(yàn)證機(jī)制
- 編輯器支持:完善的IDE自動(dòng)補(bǔ)全和類型提示
本博客將深入探討Pydantic的使用,通過(guò)理論講解和實(shí)際代碼示例,展示如何在項(xiàng)目中高效利用Pydantic進(jìn)行數(shù)據(jù)驗(yàn)證與序列化。
2. Pydantic基礎(chǔ)概念
2.1 Pydantic的核心原理
Pydantic基于Python的類型注解系統(tǒng),在運(yùn)行時(shí)驗(yàn)證數(shù)據(jù)。它使用Python的dataclasses和類型提示功能,但提供了更強(qiáng)大的驗(yàn)證和序列化能力。其核心組件是BaseModel類,所有Pydantic模型都應(yīng)繼承此類。

2.2 基本模型定義
讓我們從一個(gè)簡(jiǎn)單的例子開(kāi)始,了解如何定義Pydantic模型:
from typing import List, Optional
from datetime import datetime
from pydantic import BaseModel, Field, validator
class User(BaseModel):
"""用戶數(shù)據(jù)模型"""
id: int
username: str
email: str
age: Optional[int] = None
is_active: bool = True
created_at: datetime = Field(default_factory=datetime.now)
tags: List[str] = []
在這個(gè)例子中,我們定義了一個(gè)User模型,包含多個(gè)字段,每個(gè)字段都有明確的類型注解??蛇x字段使用Optional類型,默認(rèn)值直接在字段定義中指定。
3. 字段類型與驗(yàn)證器
3.1 內(nèi)置字段類型
Pydantic支持多種內(nèi)置字段類型,包括:
- 基本類型:
int,float,str,bool - 復(fù)雜類型:
List,Dict,Set,Tuple - 特殊類型:
EmailStr,UrlStr,IPvAnyAddress - 日期時(shí)間類型:
datetime,date,time - 自定義類型:通過(guò)繼承
pydantic.types.ConstrainedType創(chuàng)建
3.2 字段約束與驗(yàn)證
Pydantic提供了多種方式為字段添加約束:
from pydantic import BaseModel, Field, conint, constr
from typing import Optional
class Product(BaseModel):
"""產(chǎn)品數(shù)據(jù)模型"""
id: int = Field(..., gt=0, description="產(chǎn)品ID,必須大于0")
name: constr(min_length=1, max_length=100) # 字符串長(zhǎng)度約束
price: float = Field(..., gt=0, le=10000, description="價(jià)格范圍0-10000")
stock: conint(ge=0) = 0 # 整數(shù)約束,大于等于0
category: Optional[str] = Field(None, regex=r"^[A-Z][a-z]+$")
# 使用Field的更多參數(shù)
description: str = Field(
"",
max_length=500,
title="產(chǎn)品描述",
description="產(chǎn)品的詳細(xì)描述信息"
)
3.3 自定義驗(yàn)證器
除了內(nèi)置約束,還可以創(chuàng)建自定義驗(yàn)證器:
from pydantic import BaseModel, validator
from typing import List
class Order(BaseModel):
"""訂單數(shù)據(jù)模型"""
items: List[str]
quantities: List[int]
total_amount: float
@validator('quantities')
def validate_quantities(cls, v, values):
"""驗(yàn)證數(shù)量列表"""
if len(v) != len(values.get('items', [])):
raise ValueError('數(shù)量列表與商品列表長(zhǎng)度必須一致')
if any(q <= 0 for q in v):
raise ValueError('所有商品數(shù)量必須大于0')
return v
@validator('total_amount')
def validate_total_amount(cls, v, values):
"""驗(yàn)證總金額"""
quantities = values.get('quantities', [])
# 模擬計(jì)算:假設(shè)每個(gè)商品單價(jià)為10
calculated_total = sum(q * 10 for q in quantities)
if abs(v - calculated_total) > 0.01: # 允許微小誤差
raise ValueError(f'總金額計(jì)算錯(cuò)誤,應(yīng)為{calculated_total}')
return v
3.4 根驗(yàn)證器
對(duì)于需要訪問(wèn)多個(gè)字段的驗(yàn)證邏輯,可以使用根驗(yàn)證器:
from pydantic import BaseModel, root_validator
from typing import Dict, Any
class RegistrationForm(BaseModel):
"""注冊(cè)表單模型"""
username: str
password: str
confirm_password: str
email: str
@root_validator(pre=True)
def validate_all_fields_present(cls, values: Dict[str, Any]) -> Dict[str, Any]:
"""驗(yàn)證所有必填字段都存在"""
required_fields = ['username', 'password', 'confirm_password', 'email']
missing = [field for field in required_fields if field not in values]
if missing:
raise ValueError(f'缺少必填字段: {missing}')
return values
@root_validator
def validate_passwords_match(cls, values: Dict[str, Any]) -> Dict[str, Any]:
"""驗(yàn)證兩次輸入的密碼是否一致"""
password = values.get('password')
confirm_password = values.get('confirm_password')
if password and confirm_password and password != confirm_password:
raise ValueError('兩次輸入的密碼不一致')
# 密碼強(qiáng)度驗(yàn)證
if len(password) < 8:
raise ValueError('密碼長(zhǎng)度至少8位')
if not any(c.isupper() for c in password):
raise ValueError('密碼必須包含至少一個(gè)大寫字母')
if not any(c.isdigit() for c in password):
raise ValueError('密碼必須包含至少一個(gè)數(shù)字')
return values
4. 高級(jí)特性
4.1 嵌套模型
Pydantic支持復(fù)雜的嵌套模型,非常適合處理層次化數(shù)據(jù):
from typing import List, Optional
from pydantic import BaseModel, Field
class Address(BaseModel):
"""地址模型"""
street: str
city: str
state: str
zip_code: str
country: str = "中國(guó)"
class Config:
schema_extra = {
"example": {
"street": "人民路123號(hào)",
"city": "北京",
"state": "北京",
"zip_code": "100000"
}
}
class ContactInfo(BaseModel):
"""聯(lián)系信息模型"""
phone: str = Field(..., regex=r'^1[3-9]\d{9}$')
email: str
address: Address
class Company(BaseModel):
"""公司模型"""
name: str
tax_id: str = Field(..., min_length=15, max_length=20)
contacts: List[ContactInfo]
headquarters: Optional[Address] = None
def get_primary_contact(self) -> Optional[ContactInfo]:
"""獲取主要聯(lián)系人"""
return self.contacts[0] if self.contacts else None
4.2 模型繼承
Pydantic模型支持繼承,便于代碼復(fù)用:
from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional
class BaseEntity(BaseModel):
"""基礎(chǔ)實(shí)體模型"""
id: int = Field(..., gt=0)
created_at: datetime = Field(default_factory=datetime.now)
updated_at: Optional[datetime] = None
is_deleted: bool = False
class Config:
"""模型配置"""
validate_assignment = True # 啟用賦值驗(yàn)證
anystr_strip_whitespace = True # 自動(dòng)去除字符串空格
class Customer(BaseEntity):
"""客戶模型,繼承自BaseEntity"""
name: str
email: str
phone: Optional[str] = None
loyalty_points: int = Field(0, ge=0)
def add_points(self, points: int) -> None:
"""添加積分"""
if points > 0:
self.loyalty_points += points
class Config(BaseEntity.Config):
"""繼承基礎(chǔ)配置并擴(kuò)展"""
schema_extra = {
"example": {
"id": 1,
"name": "張三",
"email": "zhangsan@example.com",
"phone": "13800138000"
}
}
4.3 泛型支持
Pydantic支持泛型,可以創(chuàng)建可重用的通用模型:
from typing import TypeVar, Generic, List, Optional
from pydantic import BaseModel, Field
from pydantic.generics import GenericModel
T = TypeVar('T')
class PaginationParams(BaseModel):
"""分頁(yè)參數(shù)"""
page: int = Field(1, gt=0)
size: int = Field(10, gt=0, le=100)
class PaginatedResponse(GenericModel, Generic[T]):
"""分頁(yè)響應(yīng)泛型模型"""
items: List[T]
total: int
page: int
size: int
pages: int
@classmethod
def create(
cls,
items: List[T],
total: int,
params: PaginationParams
) -> 'PaginatedResponse[T]':
"""創(chuàng)建分頁(yè)響應(yīng)"""
pages = (total + params.size - 1) // params.size
return cls(
items=items,
total=total,
page=params.page,
size=params.size,
pages=pages
)
class ApiResponse(GenericModel, Generic[T]):
"""API響應(yīng)泛型模型"""
success: bool
data: Optional[T] = None
message: Optional[str] = None
error_code: Optional[str] = None
@classmethod
def success_response(cls, data: T) -> 'ApiResponse[T]':
"""成功響應(yīng)"""
return cls(success=True, data=data)
@classmethod
def error_response(
cls,
message: str,
error_code: str = "UNKNOWN_ERROR"
) -> 'ApiResponse[None]':
"""錯(cuò)誤響應(yīng)"""
return cls(
success=False,
message=message,
error_code=error_code
)
5. 配置與序列化
5.1 模型配置
Pydantic提供了豐富的配置選項(xiàng):
from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional
class ConfigExample(BaseModel):
"""配置示例模型"""
sensitive_data: str
normal_data: str
created_at: datetime
class Config:
# 序列化配置
json_encoders = {
datetime: lambda dt: dt.strftime('%Y-%m-%d %H:%M:%S')
}
# 字段別名
fields = {
'sensitive_data': {'exclude': True}, # 從序列化中排除
'normal_data': {'alias': 'data'} # 使用別名
}
# 驗(yàn)證配置
validate_assignment = True # 賦值時(shí)驗(yàn)證
extra = 'forbid' # 禁止額外字段
anystr_lower = True # 自動(dòng)轉(zhuǎn)換為小寫
# ORM模式
orm_mode = True
# 使用配置的示例
example = ConfigExample(
sensitive_data="secret",
normal_data="Hello World",
created_at=datetime.now()
)
# 序列化為字典(排除敏感字段)
print(example.dict(exclude={'sensitive_data'}))
5.2 序列化與反序列化
Pydantic提供了多種序列化和反序列化方法:
import json
from datetime import datetime
from typing import List
from pydantic import BaseModel, Field
class Book(BaseModel):
"""書(shū)籍模型"""
title: str
author: str
isbn: str = Field(..., regex=r'^\d{13}$')
price: float = Field(..., gt=0)
published_date: datetime
categories: List[str] = []
class Config:
json_encoders = {
datetime: lambda dt: dt.isoformat()
}
schema_extra = {
"example": {
"title": "Python編程從入門到實(shí)踐",
"author": "Eric Matthes",
"isbn": "9787115428028",
"price": 89.00,
"published_date": "2020-10-01T00:00:00",
"categories": ["編程", "Python"]
}
}
# 創(chuàng)建實(shí)例
book = Book(
title="Python高級(jí)編程",
author="Luciano Ramalho",
isbn="9787115390592",
price=99.00,
published_date=datetime(2021, 5, 1),
categories=["編程", "Python", "高級(jí)"]
)
# 序列化為字典
book_dict = book.dict()
print("字典格式:", book_dict)
# 序列化為JSON
book_json = book.json()
print("JSON格式:", book_json)
# 序列化時(shí)排除字段
book_dict_excluded = book.dict(exclude={'price'})
print("排除價(jià)格字段:", book_dict_excluded)
# 只包含特定字段
book_dict_included = book.dict(include={'title', 'author'})
print("僅包含標(biāo)題和作者:", book_dict_included)
# 反序列化
json_data = '''
{
"title": "流暢的Python",
"author": "Luciano Ramalho",
"isbn": "9787115454157",
"price": 109.00,
"published_date": "2022-03-01T00:00:00",
"categories": ["編程", "Python"]
}
'''
parsed_book = Book.parse_raw(json_data)
print("反序列化結(jié)果:", parsed_book)
# 從字典創(chuàng)建
data_dict = {
"title": "Python Cookbook",
"author": "David Beazley",
"isbn": "9781449340377",
"price": 118.00,
"published_date": "2020-08-01T00:00:00"
}
book_from_dict = Book(**data_dict)
print("從字典創(chuàng)建:", book_from_dict)
5.3 高級(jí)序列化技巧
from typing import Dict, Any, List
from pydantic import BaseModel, Field
class Product(BaseModel):
"""產(chǎn)品模型,演示高級(jí)序列化"""
id: int
name: str
price: float
inventory: int
metadata: Dict[str, Any] = Field(default_factory=dict)
def to_api_response(self) -> Dict[str, Any]:
"""轉(zhuǎn)換為API響應(yīng)格式"""
return {
"product": {
"id": self.id,
"name": self.name,
"price": self.price,
"in_stock": self.inventory > 0,
"inventory": self.inventory if self.inventory > 10 else "低庫(kù)存"
},
"metadata": self.metadata
}
@classmethod
def from_api_request(cls, data: Dict[str, Any]) -> 'Product':
"""從API請(qǐng)求數(shù)據(jù)創(chuàng)建實(shí)例"""
# 預(yù)處理數(shù)據(jù)
processed_data = data.copy()
if 'price' in processed_data:
# 確保價(jià)格是浮點(diǎn)數(shù)
processed_data['price'] = float(processed_data['price'])
return cls(**processed_data)
class ProductCatalog(BaseModel):
"""產(chǎn)品目錄"""
products: List[Product]
total_value: float = Field(0, description="庫(kù)存總價(jià)值")
@classmethod
def from_products(cls, products: List[Product]) -> 'ProductCatalog':
"""從產(chǎn)品列表創(chuàng)建目錄"""
total_value = sum(p.price * p.inventory for p in products)
return cls(products=products, total_value=total_value)
def to_summary_dict(self) -> Dict[str, Any]:
"""轉(zhuǎn)換為摘要字典"""
return {
"product_count": len(self.products),
"total_value": round(self.total_value, 2),
"average_price": round(
self.total_value / sum(p.inventory for p in self.products),
2
) if self.products else 0
}
6. 實(shí)際應(yīng)用示例
6.1 API請(qǐng)求/響應(yīng)處理
from typing import Optional, List
from datetime import datetime
from pydantic import BaseModel, Field, validator
from enum import Enum
class OrderStatus(str, Enum):
"""訂單狀態(tài)枚舉"""
PENDING = "pending"
PROCESSING = "processing"
SHIPPED = "shipped"
DELIVERED = "delivered"
CANCELLED = "cancelled"
class OrderItem(BaseModel):
"""訂單項(xiàng)"""
product_id: int = Field(..., gt=0)
quantity: int = Field(..., gt=0, le=100)
unit_price: float = Field(..., gt=0)
@property
def total_price(self) -> float:
"""計(jì)算總價(jià)"""
return self.quantity * self.unit_price
class CreateOrderRequest(BaseModel):
"""創(chuàng)建訂單請(qǐng)求"""
customer_id: int = Field(..., gt=0)
items: List[OrderItem] = Field(..., min_items=1)
shipping_address: str
notes: Optional[str] = None
@validator('items')
def validate_items(cls, v):
"""驗(yàn)證訂單項(xiàng)"""
# 檢查是否有重復(fù)的商品ID
product_ids = [item.product_id for item in v]
if len(product_ids) != len(set(product_ids)):
raise ValueError('訂單中存在重復(fù)的商品')
return v
@property
def total_amount(self) -> float:
"""計(jì)算訂單總金額"""
return sum(item.total_price for item in self.items)
class OrderResponse(BaseModel):
"""訂單響應(yīng)"""
order_id: int
customer_id: int
items: List[OrderItem]
status: OrderStatus
total_amount: float
created_at: datetime
estimated_delivery: Optional[datetime] = None
class Config:
json_encoders = {
datetime: lambda dt: dt.isoformat()
}
schema_extra = {
"example": {
"order_id": 12345,
"customer_id": 1001,
"status": "processing",
"total_amount": 299.99,
"created_at": "2024-01-15T10:30:00"
}
}
class OrderService:
"""訂單服務(wù)類"""
@staticmethod
def create_order(request: CreateOrderRequest) -> OrderResponse:
"""創(chuàng)建訂單"""
# 模擬訂單創(chuàng)建邏輯
order_id = 12345 # 實(shí)際應(yīng)從數(shù)據(jù)庫(kù)生成
# 計(jì)算預(yù)計(jì)送達(dá)時(shí)間(3天后)
from datetime import timedelta
estimated_delivery = datetime.now() + timedelta(days=3)
return OrderResponse(
order_id=order_id,
customer_id=request.customer_id,
items=request.items,
status=OrderStatus.PENDING,
total_amount=request.total_amount,
created_at=datetime.now(),
estimated_delivery=estimated_delivery
)
@staticmethod
def validate_order_data(data: dict) -> Optional[str]:
"""驗(yàn)證訂單數(shù)據(jù),返回錯(cuò)誤信息或None"""
try:
CreateOrderRequest(**data)
return None
except Exception as e:
return str(e)
6.2 數(shù)據(jù)庫(kù)模型集成
from typing import Optional, List
from datetime import datetime
from pydantic import BaseModel, Field, validator
from sqlalchemy import Column, Integer, String, Float, DateTime, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session
# SQLAlchemy基礎(chǔ)類
Base = declarative_base()
# SQLAlchemy模型
class ProductDB(Base):
"""產(chǎn)品數(shù)據(jù)庫(kù)模型"""
__tablename__ = 'products'
id = Column(Integer, primary_key=True)
name = Column(String(100), nullable=False)
description = Column(String(500))
price = Column(Float, nullable=False)
stock = Column(Integer, default=0)
category = Column(String(50))
is_active = Column(Boolean, default=True)
created_at = Column(DateTime, default=datetime.now)
updated_at = Column(DateTime, onupdate=datetime.now)
# Pydantic模型
class ProductBase(BaseModel):
"""產(chǎn)品基礎(chǔ)模型"""
name: str = Field(..., max_length=100)
description: Optional[str] = Field(None, max_length=500)
price: float = Field(..., gt=0)
stock: int = Field(0, ge=0)
category: Optional[str] = Field(None, max_length=50)
@validator('price')
def validate_price(cls, v):
"""價(jià)格驗(yàn)證"""
if v > 1000000:
raise ValueError('價(jià)格過(guò)高')
return round(v, 2)
class ProductCreate(ProductBase):
"""創(chuàng)建產(chǎn)品模型"""
pass
class ProductUpdate(BaseModel):
"""更新產(chǎn)品模型"""
name: Optional[str] = Field(None, max_length=100)
description: Optional[str] = Field(None, max_length=500)
price: Optional[float] = Field(None, gt=0)
stock: Optional[int] = Field(None, ge=0)
category: Optional[str] = Field(None, max_length=50)
is_active: Optional[bool] = None
class ProductResponse(ProductBase):
"""產(chǎn)品響應(yīng)模型"""
id: int
is_active: bool
created_at: datetime
updated_at: Optional[datetime] = None
class Config:
orm_mode = True # 啟用ORM模式
class ProductRepository:
"""產(chǎn)品倉(cāng)庫(kù)類"""
@staticmethod
def create(db: Session, product: ProductCreate) -> ProductDB:
"""創(chuàng)建產(chǎn)品"""
db_product = ProductDB(**product.dict())
db.add(db_product)
db.commit()
db.refresh(db_product)
return db_product
@staticmethod
def get(db: Session, product_id: int) -> Optional[ProductDB]:
"""獲取產(chǎn)品"""
return db.query(ProductDB).filter(
ProductDB.id == product_id,
ProductDB.is_active == True
).first()
@staticmethod
def update(
db: Session,
product_id: int,
update_data: ProductUpdate
) -> Optional[ProductDB]:
"""更新產(chǎn)品"""
db_product = ProductRepository.get(db, product_id)
if not db_product:
return None
# 更新字段
update_dict = update_data.dict(exclude_unset=True)
for key, value in update_dict.items():
setattr(db_product, key, value)
db_product.updated_at = datetime.now()
db.commit()
db.refresh(db_product)
return db_product
@staticmethod
def to_pydantic(db_product: ProductDB) -> ProductResponse:
"""轉(zhuǎn)換為Pydantic響應(yīng)模型"""
return ProductResponse.from_orm(db_product)
7. 完整代碼示例
"""
Pydantic數(shù)據(jù)驗(yàn)證與序列化完整示例
演示用戶管理系統(tǒng)中的數(shù)據(jù)處理
"""
import json
from datetime import datetime, date
from typing import List, Optional, Dict, Any
from enum import Enum
from uuid import uuid4
from pydantic import (
BaseModel,
Field,
validator,
root_validator,
EmailStr,
HttpUrl
)
from pydantic.generics import GenericModel
from typing import Generic, TypeVar
# 定義泛型類型
T = TypeVar('T')
# 枚舉定義
class UserRole(str, Enum):
"""用戶角色枚舉"""
ADMIN = "admin"
USER = "user"
GUEST = "guest"
MODERATOR = "moderator"
class AccountStatus(str, Enum):
"""賬戶狀態(tài)枚舉"""
ACTIVE = "active"
INACTIVE = "inactive"
SUSPENDED = "suspended"
BANNED = "banned"
# 基礎(chǔ)模型
class TimestampMixin(BaseModel):
"""時(shí)間戳混合類"""
created_at: datetime = Field(default_factory=datetime.now)
updated_at: Optional[datetime] = None
class Config:
validate_assignment = True
# 地址模型
class Address(BaseModel):
"""地址信息"""
street: str = Field(..., max_length=200)
city: str = Field(..., max_length=100)
state: str = Field(..., max_length=50)
zip_code: str = Field(..., regex=r'^\d{6}$')
country: str = "中國(guó)"
@property
def full_address(self) -> str:
"""獲取完整地址"""
return f"{self.country}{self.state}{self.city}{self.street},郵編:{self.zip_code}"
# 聯(lián)系信息模型
class ContactInfo(BaseModel):
"""聯(lián)系信息"""
email: EmailStr
phone: Optional[str] = Field(None, regex=r'^1[3-9]\d{9}$')
website: Optional[HttpUrl] = None
@validator('phone')
def validate_phone(cls, v):
"""驗(yàn)證手機(jī)號(hào)"""
if v and not v.startswith('1'):
raise ValueError('手機(jī)號(hào)格式不正確')
return v
# 用戶基礎(chǔ)模型
class UserBase(TimestampMixin):
"""用戶基礎(chǔ)信息"""
username: str = Field(
...,
min_length=3,
max_length=50,
regex=r'^[a-zA-Z][a-zA-Z0-9_]*$',
description="用戶名,只能包含字母、數(shù)字和下劃線"
)
display_name: str = Field(..., max_length=100)
email: EmailStr
birth_date: Optional[date] = None
role: UserRole = UserRole.USER
status: AccountStatus = AccountStatus.ACTIVE
@validator('birth_date')
def validate_birth_date(cls, v):
"""驗(yàn)證出生日期"""
if v:
if v > date.today():
raise ValueError('出生日期不能在未來(lái)')
# 檢查年齡是否合理(假設(shè)用戶年齡在0-150歲之間)
age = (date.today() - v).days // 365
if age > 150:
raise ValueError('年齡不合理')
return v
# 用戶創(chuàng)建模型
class UserCreate(UserBase):
"""創(chuàng)建用戶模型"""
password: str = Field(..., min_length=8)
confirm_password: str
@root_validator
def validate_passwords(cls, values):
"""驗(yàn)證密碼"""
password = values.get('password')
confirm_password = values.get('confirm_password')
if password and confirm_password and password != confirm_password:
raise ValueError('兩次輸入的密碼不一致')
# 密碼強(qiáng)度檢查
if password:
if not any(c.isupper() for c in password):
raise ValueError('密碼必須包含至少一個(gè)大寫字母')
if not any(c.isdigit() for c in password):
raise ValueError('密碼必須包含至少一個(gè)數(shù)字')
if not any(c in '!@#$%^&*()_+-=[]{}|;:,.<>?`~' for c in password):
raise ValueError('密碼必須包含至少一個(gè)特殊字符')
return values
# 用戶更新模型
class UserUpdate(BaseModel):
"""更新用戶模型"""
display_name: Optional[str] = Field(None, max_length=100)
email: Optional[EmailStr] = None
birth_date: Optional[date] = None
role: Optional[UserRole] = None
status: Optional[AccountStatus] = None
class Config:
extra = 'forbid' # 禁止額外字段
# 用戶完整模型
class User(UserBase):
"""用戶完整模型"""
id: str = Field(default_factory=lambda: str(uuid4()))
addresses: List[Address] = []
contact_info: ContactInfo
metadata: Dict[str, Any] = Field(default_factory=dict)
last_login: Optional[datetime] = None
login_count: int = 0
@property
def age(self) -> Optional[int]:
"""計(jì)算年齡"""
if self.birth_date:
today = date.today()
return today.year - self.birth_date.year - (
(today.month, today.day) < (self.birth_date.month, self.birth_date.day)
)
return None
def to_summary_dict(self) -> Dict[str, Any]:
"""轉(zhuǎn)換為摘要字典"""
return {
'id': self.id,
'username': self.username,
'display_name': self.display_name,
'email': self.email,
'role': self.role,
'status': self.status,
'age': self.age,
'address_count': len(self.addresses)
}
def record_login(self) -> None:
"""記錄登錄"""
self.last_login = datetime.now()
self.login_count += 1
self.updated_at = datetime.now()
# API響應(yīng)模型
class ApiResponse(GenericModel, Generic[T]):
"""通用API響應(yīng)"""
success: bool
data: Optional[T] = None
message: Optional[str] = None
error_code: Optional[str] = None
timestamp: datetime = Field(default_factory=datetime.now)
class Config:
json_encoders = {
datetime: lambda dt: dt.isoformat()
}
@classmethod
def success(cls, data: T, message: str = "操作成功") -> 'ApiResponse[T]':
"""成功響應(yīng)"""
return cls(success=True, data=data, message=message)
@classmethod
def error(
cls,
message: str,
error_code: str = "INTERNAL_ERROR"
) -> 'ApiResponse[None]':
"""錯(cuò)誤響應(yīng)"""
return cls(success=False, message=message, error_code=error_code)
# 分頁(yè)模型
class PaginationParams(BaseModel):
"""分頁(yè)參數(shù)"""
page: int = Field(1, gt=0)
size: int = Field(10, gt=0, le=100)
sort_by: Optional[str] = None
sort_order: Optional[str] = Field(None, regex=r'^(asc|desc)$')
class PaginatedResponse(GenericModel, Generic[T]):
"""分頁(yè)響應(yīng)"""
items: List[T]
total: int
page: int
size: int
pages: int
has_next: bool
has_prev: bool
@classmethod
def create(
cls,
items: List[T],
total: int,
params: PaginationParams
) -> 'PaginatedResponse[T]':
"""創(chuàng)建分頁(yè)響應(yīng)"""
pages = (total + params.size - 1) // params.size
has_next = params.page < pages
has_prev = params.page > 1
return cls(
items=items,
total=total,
page=params.page,
size=params.size,
pages=pages,
has_next=has_next,
has_prev=has_prev
)
# 用戶服務(wù)類
class UserService:
"""用戶服務(wù)"""
def __init__(self):
self.users: Dict[str, User] = {}
def create_user(self, user_data: UserCreate) -> ApiResponse[User]:
"""創(chuàng)建用戶"""
try:
# 檢查用戶名是否已存在
if any(u.username == user_data.username for u in self.users.values()):
return ApiResponse.error("用戶名已存在", "USERNAME_EXISTS")
# 檢查郵箱是否已存在
if any(u.email == user_data.email for u in self.users.values()):
return ApiResponse.error("郵箱已存在", "EMAIL_EXISTS")
# 創(chuàng)建用戶(排除密碼字段)
user_dict = user_data.dict(exclude={'password', 'confirm_password'})
user = User(**user_dict)
# 添加聯(lián)系信息(示例)
user.contact_info = ContactInfo(email=user_data.email)
# 保存用戶
self.users[user.id] = user
return ApiResponse.success(user, "用戶創(chuàng)建成功")
except Exception as e:
return ApiResponse.error(f"創(chuàng)建用戶失敗: {str(e)}")
def get_user(self, user_id: str) -> ApiResponse[User]:
"""獲取用戶"""
user = self.users.get(user_id)
if not user:
return ApiResponse.error("用戶不存在", "USER_NOT_FOUND")
return ApiResponse.success(user)
def update_user(
self,
user_id: str,
update_data: UserUpdate
) -> ApiResponse[User]:
"""更新用戶"""
user = self.users.get(user_id)
if not user:
return ApiResponse.error("用戶不存在", "USER_NOT_FOUND")
try:
# 更新字段
update_dict = update_data.dict(exclude_unset=True)
for key, value in update_dict.items():
setattr(user, key, value)
user.updated_at = datetime.now()
return ApiResponse.success(user, "用戶更新成功")
except Exception as e:
return ApiResponse.error(f"更新用戶失敗: {str(e)}")
def list_users(
self,
params: PaginationParams
) -> ApiResponse[PaginatedResponse[User]]:
"""用戶列表"""
try:
# 獲取所有用戶
all_users = list(self.users.values())
# 排序
if params.sort_by:
reverse = params.sort_order == 'desc'
all_users.sort(
key=lambda u: getattr(u, params.sort_by, u.username),
reverse=reverse
)
# 分頁(yè)
start = (params.page - 1) * params.size
end = start + params.size
paginated_users = all_users[start:end]
# 創(chuàng)建分頁(yè)響應(yīng)
paginated_response = PaginatedResponse.create(
items=paginated_users,
total=len(all_users),
params=params
)
return ApiResponse.success(paginated_response)
except Exception as e:
return ApiResponse.error(f"獲取用戶列表失敗: {str(e)}")
# 演示函數(shù)
def demonstrate_pydantic_features():
"""演示Pydantic功能"""
print("=" * 60)
print("Pydantic數(shù)據(jù)驗(yàn)證與序列化演示")
print("=" * 60)
# 1. 創(chuàng)建用戶
print("\n1. 創(chuàng)建用戶")
user_service = UserService()
# 正確的用戶數(shù)據(jù)
valid_user_data = {
"username": "john_doe",
"display_name": "John Doe",
"email": "john@example.com",
"password": "SecurePass123!",
"confirm_password": "SecurePass123!",
"birth_date": "1990-01-01",
"role": "user",
"status": "active"
}
create_response = user_service.create_user(UserCreate(**valid_user_data))
if create_response.success:
print(f"用戶創(chuàng)建成功: {create_response.data.username}")
# 2. 獲取用戶
print("\n2. 獲取用戶")
user_id = create_response.data.id
get_response = user_service.get_user(user_id)
if get_response.success:
user = get_response.data
print(f"用戶信息: {user.to_summary_dict()}")
# 3. 序列化為JSON
print("\n3. 序列化為JSON")
user_json = user.json(indent=2)
print("用戶JSON表示:")
print(user_json)
# 4. 反序列化
print("\n4. 反序列化")
parsed_user = User.parse_raw(user_json)
print(f"反序列化成功: {parsed_user.username}")
# 5. 更新用戶
print("\n5. 更新用戶")
update_data = UserUpdate(
display_name="John Smith",
role=UserRole.MODERATOR
)
update_response = user_service.update_user(user_id, update_data)
if update_response.success:
print(f"用戶更新成功: {update_response.data.display_name}")
# 6. 用戶列表分頁(yè)
print("\n6. 用戶列表分頁(yè)")
pagination_params = PaginationParams(page=1, size=5)
list_response = user_service.list_users(pagination_params)
if list_response.success:
paginated_data = list_response.data
print(f"總用戶數(shù): {paginated_data.total}")
print(f"當(dāng)前頁(yè): {paginated_data.page}/{paginated_data.pages}")
print(f"每頁(yè)大小: {paginated_data.size}")
# 7. 錯(cuò)誤處理演示
print("\n7. 錯(cuò)誤處理演示")
# 無(wú)效的用戶名
print("\n嘗試使用無(wú)效用戶名:")
invalid_username_data = valid_user_data.copy()
invalid_username_data['username'] = "123invalid" # 以數(shù)字開(kāi)頭
try:
UserCreate(**invalid_username_data)
except Exception as e:
print(f"錯(cuò)誤: {e}")
# 密碼不匹配
print("\n嘗試使用不匹配的密碼:")
invalid_password_data = valid_user_data.copy()
invalid_password_data['confirm_password'] = "DifferentPass123!"
try:
UserCreate(**invalid_password_data)
except Exception as e:
print(f"錯(cuò)誤: {e}")
# 弱密碼
print("\n嘗試使用弱密碼:")
weak_password_data = valid_user_data.copy()
weak_password_data['password'] = weak_password_data['confirm_password'] = "weak"
try:
UserCreate(**weak_password_data)
except Exception as e:
print(f"錯(cuò)誤: {e}")
print("\n" + "=" * 60)
print("演示完成")
print("=" * 60)
if __name__ == "__main__":
# 運(yùn)行演示
demonstrate_pydantic_features()
8. 代碼自查與優(yōu)化
為確保代碼質(zhì)量,我們進(jìn)行了以下自查和優(yōu)化:
8.1 代碼自查清單
- 類型注解完整性檢查:所有函數(shù)參數(shù)和返回值都有明確的類型注解
- 異常處理:關(guān)鍵操作都有適當(dāng)?shù)漠惓L幚頇C(jī)制
- 輸入驗(yàn)證:所有用戶輸入都經(jīng)過(guò)Pydantic驗(yàn)證
- 代碼可讀性:使用清晰的變量名和函數(shù)名,添加必要的注釋
- 性能考慮:避免在循環(huán)中進(jìn)行重復(fù)驗(yàn)證,使用適當(dāng)?shù)臄?shù)據(jù)結(jié)構(gòu)
- 安全性:敏感字段(如密碼)在序列化時(shí)被排除
- 錯(cuò)誤消息:提供清晰、有用的錯(cuò)誤消息
- 測(cè)試覆蓋:示例代碼包含主要功能的演示
8.2 常見(jiàn)問(wèn)題與解決方案
- 循環(huán)引用問(wèn)題:使用
ForwardRef或字符串類型注解 - 性能優(yōu)化:對(duì)于大量數(shù)據(jù),考慮使用
parse_obj_as進(jìn)行批量解析 - 自定義驗(yàn)證:復(fù)雜的驗(yàn)證邏輯拆分為多個(gè)驗(yàn)證器
- 配置管理:敏感配置通過(guò)環(huán)境變量或配置文件管理
- 版本兼容性:注意Pydantic版本差異,特別是v1和v2之間的變化
8.3 最佳實(shí)踐建議
模型設(shè)計(jì):
- 保持模型職責(zé)單一
- 使用繼承減少重復(fù)代碼
- 為常用操作創(chuàng)建便捷方法
驗(yàn)證策略:
- 在數(shù)據(jù)入口處進(jìn)行驗(yàn)證
- 使用細(xì)粒度的驗(yàn)證器
- 提供有意義的錯(cuò)誤消息
序列化優(yōu)化:
- 使用
exclude和include參數(shù)控制輸出字段 - 為不同場(chǎng)景創(chuàng)建不同的響應(yīng)模型
- 使用自定義JSON編碼器處理特殊類型
9. 總結(jié)
Pydantic作為現(xiàn)代Python生態(tài)系統(tǒng)中數(shù)據(jù)驗(yàn)證和序列化的首選工具,提供了強(qiáng)大而靈活的功能。通過(guò)本文的詳細(xì)介紹和代碼示例,我們看到了Pydantic如何:
- 提高代碼可靠性:通過(guò)運(yùn)行時(shí)類型檢查減少錯(cuò)誤
- 簡(jiǎn)化數(shù)據(jù)處理:提供直觀的API進(jìn)行數(shù)據(jù)驗(yàn)證和轉(zhuǎn)換
- 增強(qiáng)開(kāi)發(fā)體驗(yàn):完善的IDE支持和類型提示
- 促進(jìn)代碼重用:通過(guò)模型繼承和泛型支持提高代碼復(fù)用率
9.1 適用場(chǎng)景
Pydantic特別適用于以下場(chǎng)景:
- API開(kāi)發(fā):請(qǐng)求/響應(yīng)數(shù)據(jù)的驗(yàn)證和序列化
- 配置管理:應(yīng)用程序配置的加載和驗(yàn)證
- 數(shù)據(jù)管道:數(shù)據(jù)清洗和轉(zhuǎn)換過(guò)程中的驗(yàn)證
- 數(shù)據(jù)庫(kù)交互:ORM模型與業(yè)務(wù)模型之間的轉(zhuǎn)換
9.2 未來(lái)展望
隨著Python類型系統(tǒng)的不斷完善和Pydantic社區(qū)的持續(xù)發(fā)展,我們可以期待更多高級(jí)功能的加入,如:
- 更強(qiáng)大的自定義類型系統(tǒng)
- 性能優(yōu)化的驗(yàn)證機(jī)制
- 更好的異步支持
- 更豐富的生態(tài)系統(tǒng)集成
通過(guò)合理使用Pydantic,我們可以構(gòu)建更加健壯、可維護(hù)的Python應(yīng)用程序,有效減少數(shù)據(jù)相關(guān)的錯(cuò)誤,提高開(kāi)發(fā)效率。希望本文能幫助您更好地理解和應(yīng)用Pydantic,在您的項(xiàng)目中發(fā)揮其最大價(jià)值。
注意:本文代碼示例已在Python 3.8+和Pydantic 2.0+環(huán)境下測(cè)試通過(guò)。在實(shí)際使用中,請(qǐng)根據(jù)具體需求調(diào)整代碼,并添加適當(dāng)?shù)腻e(cuò)誤處理和日志記錄。
以上就是Python使用Pydantic進(jìn)行數(shù)據(jù)驗(yàn)證與序列化詳解的詳細(xì)內(nèi)容,更多關(guān)于Python Pydantic使用的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
使用pycharm和pylint檢查python代碼規(guī)范操作
這篇文章主要介紹了使用pycharm和pylint檢查python代碼規(guī)范操作,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2020-06-06
python模擬登錄百度代碼分享(獲取百度貼吧等級(jí))
python模擬登錄百度,實(shí)現(xiàn)了登錄并讀取百度關(guān)注貼吧列表,百度登錄還是有點(diǎn)麻煩的,由于用的ssl,所以要先獲取token,然后再登錄,這個(gè)用finddle2分析下,還是比較好解決的2013-12-12
Python實(shí)現(xiàn)簡(jiǎn)單求解給定整數(shù)的質(zhì)因數(shù)算法示例
這篇文章主要介紹了Python實(shí)現(xiàn)簡(jiǎn)單求解給定整數(shù)的質(zhì)因數(shù)算法,結(jié)合實(shí)例形式分析了Python正整數(shù)分解質(zhì)因數(shù)的相關(guān)操作技巧,需要的朋友可以參考下2018-03-03
Python 網(wǎng)絡(luò)編程之UDP發(fā)送接收數(shù)據(jù)功能示例【基于socket套接字】
這篇文章主要介紹了Python 網(wǎng)絡(luò)編程之UDP發(fā)送接收數(shù)據(jù)功能,結(jié)合實(shí)例形式分析了Python使用socket套接字實(shí)現(xiàn)基于UDP協(xié)議的數(shù)據(jù)發(fā)送端與接收端相關(guān)操作技巧,需要的朋友可以參考下2019-10-10
python3安裝及pip3報(bào)ERROR:No?matching?distribution?found?for解
這篇文章主要給大家介紹了關(guān)于python3安裝及pip3報(bào)ERROR:No?matching?distribution?found?for解決的相關(guān)資料,文中通過(guò)代碼以及圖文介紹的非常詳細(xì),需要的朋友可以參考下2023-08-08
PyQt6/PySide6中QTableView類的實(shí)現(xiàn)
本文主要介紹了PyQt6/PySide6中QTableView類的實(shí)現(xiàn),文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2025-04-04

