Python使用Pydantic进行数据验证与序列化详解

from typing import list, optional
from datetime import datetime
from pydantic import basemodel, field, validator

class user(basemodel):
    """用户数据模型"""
    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] = []

from pydantic import basemodel, field, conint, constr
from typing import optional

class product(basemodel):
    """产品数据模型"""
    id: int = field(..., gt=0, description="产品id，必须大于0")
    name: constr(min_length=1, max_length=100)  # 字符串长度约束
    price: float = field(..., gt=0, le=10000, description="价格范围0-10000")
    stock: conint(ge=0) = 0  # 整数约束，大于等于0
    category: optional[str] = field(none, regex=r"^[a-z][a-z]+$")
    
    # 使用field的更多参数
    description: str = field(
        "",
        max_length=500,
        title="产品描述",
        description="产品的详细描述信息"
    )

from pydantic import basemodel, validator
from typing import list

class order(basemodel):
    """订单数据模型"""
    items: list[str]
    quantities: list[int]
    total_amount: float
    
    @validator('quantities')
    def validate_quantities(cls, v, values):
        """验证数量列表"""
        if len(v) != len(values.get('items', [])):
            raise valueerror('数量列表与商品列表长度必须一致')
        if any(q <= 0 for q in v):
            raise valueerror('所有商品数量必须大于0')
        return v
    
    @validator('total_amount')
    def validate_total_amount(cls, v, values):
        """验证总金额"""
        quantities = values.get('quantities', [])
        # 模拟计算：假设每个商品单价为10
        calculated_total = sum(q * 10 for q in quantities)
        if abs(v - calculated_total) > 0.01:  # 允许微小误差
            raise valueerror(f'总金额计算错误，应为{calculated_total}')
        return v

from pydantic import basemodel, root_validator
from typing import dict, any

class registrationform(basemodel):
    """注册表单模型"""
    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]:
        """验证所有必填字段都存在"""
        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]:
        """验证两次输入的密码是否一致"""
        password = values.get('password')
        confirm_password = values.get('confirm_password')
        
        if password and confirm_password and password != confirm_password:
            raise valueerror('两次输入的密码不一致')
        
        # 密码强度验证
        if len(password) < 8:
            raise valueerror('密码长度至少8位')
        if not any(c.isupper() for c in password):
            raise valueerror('密码必须包含至少一个大写字母')
        if not any(c.isdigit() for c in password):
            raise valueerror('密码必须包含至少一个数字')
        
        return values

from typing import list, optional
from pydantic import basemodel, field

class address(basemodel):
    """地址模型"""
    street: str
    city: str
    state: str
    zip_code: str
    country: str = "中国"
    
    class config:
        schema_extra = {
            "example": {
                "street": "人民路123号",
                "city": "北京",
                "state": "北京",
                "zip_code": "100000"
            }
        }

class contactinfo(basemodel):
    """联系信息模型"""
    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]:
        """获取主要联系人"""
        return self.contacts[0] if self.contacts else none

from pydantic import basemodel, field
from datetime import datetime
from typing import optional

class baseentity(basemodel):
    """基础实体模型"""
    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  # 启用赋值验证
        anystr_strip_whitespace = true  # 自动去除字符串空格

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):
        """继承基础配置并扩展"""
        schema_extra = {
            "example": {
                "id": 1,
                "name": "张三",
                "email": "zhangsan@example.com",
                "phone": "13800138000"
            }
        }

from typing import typevar, generic, list, optional
from pydantic import basemodel, field
from pydantic.generics import genericmodel

t = typevar('t')

class paginationparams(basemodel):
    """分页参数"""
    page: int = field(1, gt=0)
    size: int = field(10, gt=0, le=100)

class paginatedresponse(genericmodel, generic[t]):
    """分页响应泛型模型"""
    items: list[t]
    total: int
    page: int
    size: int
    pages: int
    
    @classmethod
    def create(
        cls, 
        items: list[t], 
        total: int, 
        params: paginationparams
    ) -> 'paginatedresponse[t]':
        """创建分页响应"""
        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响应泛型模型"""
    success: bool
    data: optional[t] = none
    message: optional[str] = none
    error_code: optional[str] = none
    
    @classmethod
    def success_response(cls, data: t) -> 'apiresponse[t]':
        """成功响应"""
        return cls(success=true, data=data)
    
    @classmethod
    def error_response(
        cls, 
        message: str, 
        error_code: str = "unknown_error"
    ) -> 'apiresponse[none]':
        """错误响应"""
        return cls(
            success=false, 
            message=message, 
            error_code=error_code
        )

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'}  # 使用别名
        }
        
        # 验证配置
        validate_assignment = true  # 赋值时验证
        extra = 'forbid'  # 禁止额外字段
        anystr_lower = true  # 自动转换为小写
        
        # orm模式
        orm_mode = true

# 使用配置的示例
example = configexample(
    sensitive_data="secret",
    normal_data="hello world",
    created_at=datetime.now()
)

# 序列化为字典（排除敏感字段）
print(example.dict(exclude={'sensitive_data'}))

import json
from datetime import datetime
from typing import list
from pydantic import basemodel, field

class book(basemodel):
    """书籍模型"""
    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编程从入门到实践",
                "author": "eric matthes",
                "isbn": "9787115428028",
                "price": 89.00,
                "published_date": "2020-10-01t00:00:00",
                "categories": ["编程", "python"]
            }
        }

# 创建实例
book = book(
    title="python高级编程",
    author="luciano ramalho",
    isbn="9787115390592",
    price=99.00,
    published_date=datetime(2021, 5, 1),
    categories=["编程", "python", "高级"]
)

# 序列化为字典
book_dict = book.dict()
print("字典格式:", book_dict)

# 序列化为json
book_json = book.json()
print("json格式:", book_json)

# 序列化时排除字段
book_dict_excluded = book.dict(exclude={'price'})
print("排除价格字段:", book_dict_excluded)

# 只包含特定字段
book_dict_included = book.dict(include={'title', 'author'})
print("仅包含标题和作者:", 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("反序列化结果:", parsed_book)

# 从字典创建
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("从字典创建:", book_from_dict)

from typing import dict, any, list
from pydantic import basemodel, field

class product(basemodel):
    """产品模型，演示高级序列化"""
    id: int
    name: str
    price: float
    inventory: int
    metadata: dict[str, any] = field(default_factory=dict)
    
    def to_api_response(self) -> dict[str, any]:
        """转换为api响应格式"""
        return {
            "product": {
                "id": self.id,
                "name": self.name,
                "price": self.price,
                "in_stock": self.inventory > 0,
                "inventory": self.inventory if self.inventory > 10 else "低库存"
            },
            "metadata": self.metadata
        }
    
    @classmethod
    def from_api_request(cls, data: dict[str, any]) -> 'product':
        """从api请求数据创建实例"""
        # 预处理数据
        processed_data = data.copy()
        if 'price' in processed_data:
            # 确保价格是浮点数
            processed_data['price'] = float(processed_data['price'])
        
        return cls(**processed_data)

class productcatalog(basemodel):
    """产品目录"""
    products: list[product]
    total_value: float = field(0, description="库存总价值")
    
    @classmethod
    def from_products(cls, products: list[product]) -> 'productcatalog':
        """从产品列表创建目录"""
        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]:
        """转换为摘要字典"""
        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
        }

from typing import optional, list
from datetime import datetime
from pydantic import basemodel, field, validator
from enum import enum

class orderstatus(str, enum):
    """订单状态枚举"""
    pending = "pending"
    processing = "processing"
    shipped = "shipped"
    delivered = "delivered"
    cancelled = "cancelled"

class orderitem(basemodel):
    """订单项"""
    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:
        """计算总价"""
        return self.quantity * self.unit_price

class createorderrequest(basemodel):
    """创建订单请求"""
    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):
        """验证订单项"""
        # 检查是否有重复的商品id
        product_ids = [item.product_id for item in v]
        if len(product_ids) != len(set(product_ids)):
            raise valueerror('订单中存在重复的商品')
        return v
    
    @property
    def total_amount(self) -> float:
        """计算订单总金额"""
        return sum(item.total_price for item in self.items)

class orderresponse(basemodel):
    """订单响应"""
    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:
    """订单服务类"""
    
    @staticmethod
    def create_order(request: createorderrequest) -> orderresponse:
        """创建订单"""
        # 模拟订单创建逻辑
        order_id = 12345  # 实际应从数据库生成
        
        # 计算预计送达时间（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]:
        """验证订单数据，返回错误信息或none"""
        try:
            createorderrequest(**data)
            return none
        except exception as e:
            return str(e)

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基础类
base = declarative_base()

# sqlalchemy模型
class productdb(base):
    """产品数据库模型"""
    __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):
    """产品基础模型"""
    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):
        """价格验证"""
        if v > 1000000:
            raise valueerror('价格过高')
        return round(v, 2)

class productcreate(productbase):
    """创建产品模型"""
    pass

class productupdate(basemodel):
    """更新产品模型"""
    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):
    """产品响应模型"""
    id: int
    is_active: bool
    created_at: datetime
    updated_at: optional[datetime] = none
    
    class config:
        orm_mode = true  # 启用orm模式

class productrepository:
    """产品仓库类"""
    
    @staticmethod
    def create(db: session, product: productcreate) -> productdb:
        """创建产品"""
        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]:
        """获取产品"""
        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]:
        """更新产品"""
        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:
        """转换为pydantic响应模型"""
        return productresponse.from_orm(db_product)

"""
pydantic数据验证与序列化完整示例
演示用户管理系统中的数据处理
"""

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):
    """账户状态枚举"""
    active = "active"
    inactive = "inactive"
    suspended = "suspended"
    banned = "banned"

# 基础模型
class timestampmixin(basemodel):
    """时间戳混合类"""
    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 = "中国"
    
    @property
    def full_address(self) -> str:
        """获取完整地址"""
        return f"{self.country}{self.state}{self.city}{self.street}，邮编：{self.zip_code}"

# 联系信息模型
class contactinfo(basemodel):
    """联系信息"""
    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):
        """验证手机号"""
        if v and not v.startswith('1'):
            raise valueerror('手机号格式不正确')
        return v

# 用户基础模型
class userbase(timestampmixin):
    """用户基础信息"""
    username: str = field(
        ..., 
        min_length=3, 
        max_length=50,
        regex=r'^[a-za-z][a-za-z0-9_]*$',
        description="用户名，只能包含字母、数字和下划线"
    )
    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):
        """验证出生日期"""
        if v:
            if v > date.today():
                raise valueerror('出生日期不能在未来')
            # 检查年龄是否合理（假设用户年龄在0-150岁之间）
            age = (date.today() - v).days // 365
            if age > 150:
                raise valueerror('年龄不合理')
        return v

# 用户创建模型
class usercreate(userbase):
    """创建用户模型"""
    password: str = field(..., min_length=8)
    confirm_password: str
    
    @root_validator
    def validate_passwords(cls, values):
        """验证密码"""
        password = values.get('password')
        confirm_password = values.get('confirm_password')
        
        if password and confirm_password and password != confirm_password:
            raise valueerror('两次输入的密码不一致')
        
        # 密码强度检查
        if password:
            if not any(c.isupper() for c in password):
                raise valueerror('密码必须包含至少一个大写字母')
            if not any(c.isdigit() for c in password):
                raise valueerror('密码必须包含至少一个数字')
            if not any(c in '!@#$%^&*()_+-=[]{}|;:,.<>?`~' for c in password):
                raise valueerror('密码必须包含至少一个特殊字符')
        
        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]:
        """计算年龄"""
        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]:
        """转换为摘要字典"""
        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响应模型
class apiresponse(genericmodel, generic[t]):
    """通用api响应"""
    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]':
        """成功响应"""
        return cls(success=true, data=data, message=message)
    
    @classmethod
    def error(
        cls, 
        message: str, 
        error_code: str = "internal_error"
    ) -> 'apiresponse[none]':
        """错误响应"""
        return cls(success=false, message=message, error_code=error_code)

# 分页模型
class paginationparams(basemodel):
    """分页参数"""
    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]):
    """分页响应"""
    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]':
        """创建分页响应"""
        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
        )

# 用户服务类
class userservice:
    """用户服务"""
    
    def __init__(self):
        self.users: dict[str, user] = {}
    
    def create_user(self, user_data: usercreate) -> apiresponse[user]:
        """创建用户"""
        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")
            
            # 创建用户（排除密码字段）
            user_dict = user_data.dict(exclude={'password', 'confirm_password'})
            user = user(**user_dict)
            
            # 添加联系信息（示例）
            user.contact_info = contactinfo(email=user_data.email)
            
            # 保存用户
            self.users[user.id] = user
            
            return apiresponse.success(user, "用户创建成功")
            
        except exception as e:
            return apiresponse.error(f"创建用户失败: {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
                )
            
            # 分页
            start = (params.page - 1) * params.size
            end = start + params.size
            paginated_users = all_users[start:end]
            
            # 创建分页响应
            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)}")

# 演示函数
def demonstrate_pydantic_features():
    """演示pydantic功能"""
    print("=" * 60)
    print("pydantic数据验证与序列化演示")
    print("=" * 60)
    
    # 1. 创建用户
    print("\n1. 创建用户")
    user_service = userservice()
    
    # 正确的用户数据
    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"用户创建成功: {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. 用户列表分页
            print("\n6. 用户列表分页")
            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"总用户数: {paginated_data.total}")
                print(f"当前页: {paginated_data.page}/{paginated_data.pages}")
                print(f"每页大小: {paginated_data.size}")
                
        # 7. 错误处理演示
        print("\n7. 错误处理演示")
        
        # 无效的用户名
        print("\n尝试使用无效用户名:")
        invalid_username_data = valid_user_data.copy()
        invalid_username_data['username'] = "123invalid"  # 以数字开头
        
        try:
            usercreate(**invalid_username_data)
        except exception as e:
            print(f"错误: {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"错误: {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"错误: {e}")
    
    print("\n" + "=" * 60)
    print("演示完成")
    print("=" * 60)

if __name__ == "__main__":
    # 运行演示
    demonstrate_pydantic_features()