Pydantic — 运行时数据校验与序列化

# Python 3.10+
from pydantic import BaseModel, ValidationError
from datetime import datetime


class User(BaseModel):
    """用户模型"""
    id: int
    name: str
    email: str
    age: int = 18                         # 带默认值的字段
    created_at: datetime = datetime.now()  # 默认值可以是动态值


# 正常实例化 — 自动类型转换
user = User(id=1, name="Alice", email="alice@example.com", age="25")
print(user)          # id=1 name='Alice' email='alice@example.com' age=25
print(user.age)      # 25（str "25" 自动转为 int）

# 类型校验失败
try:
    User(id="not_a_number", name="Bob", email="bob@example.com")
except ValidationError as e:
    # 错误信息结构化，包含位置和原因
    print(e.errors())
    # [{'type': 'int_parsing', 'loc': ('id',), 'msg': '...', 'input': 'not_a_number'}]

# 缺失必填字段
try:
    User(name="Charlie", email="charlie@example.com")
except ValidationError as e:
    print(f"缺失字段: {e.errors()[0]['loc']}")  # 缺失字段: ('id',)

# Python 3.10+
from pydantic import BaseModel
from datetime import datetime


class Post(BaseModel):
    id: int
    title: str
    content: str
    published_at: datetime


post = Post(
    id=1,
    title="Hello Pydantic",
    content="Pydantic makes data validation easy.",
    published_at="2026-06-05T10:00:00"
)

# 序列化为 dict（v2 新 API）
print(post.model_dump())
# {'id': 1, 'title': 'Hello Pydantic', 'content': '...', 'published_at': datetime(...)}

# 序列化为 JSON 字符串
print(post.model_dump_json(indent=2))
# {
#   "id": 1,
#   "title": "Hello Pydantic",
#   "content": "Pydantic makes data validation easy.",
#   "published_at": "2026-06-05T10:00:00"
# }

# 从 JSON 反序列化
json_data = '{"id": 2, "title": "JSON input", "content": "From JSON", "published_at": "2026-06-05T12:00:00"}'
post2 = Post.model_validate_json(json_data)
print(post2.title)  # JSON input

# Python 3.10+
from pydantic import BaseModel, Field
from typing import Optional


class Product(BaseModel):
    name: str = Field(
        ...,
        min_length=1,
        max_length=100,
        description="产品名称",
        examples=["iPhone 15"]
    )
    price: float = Field(gt=0, le=999999, description="价格，单位：元")
    quantity: int = Field(ge=0, default=0, description="库存数量")
    tags: list[str] = Field(default_factory=list, description="标签列表")
    description: Optional[str] = Field(
        default=None,
        max_length=1000,
        description="产品描述（可选）"
    )


# 正常实例化
product = Product(name="MacBook Pro", price=14999.0, quantity=50)
print(product.model_dump())

# 校验失败示例
try:
    Product(name="", price=-100, quantity=10)
except ValidationError as e:
    for err in e.errors():
        print(f"字段 {err['loc']}: {err['msg']}")

# Python 3.10+
from pydantic import BaseModel, Field, field_validator
from typing import ClassVar


class UserRegistration(BaseModel):
    username: str = Field(min_length=3, max_length=20)
    password: str = Field(min_length=8)
    password_confirm: str
    email: str

    # 禁止的用户名黑名单
    FORBIDDEN_USERNAMES: ClassVar[set[str]] = {"admin", "root", "system"}

    @field_validator("username")
    @classmethod
    def username_must_be_valid(cls, v: str) -> str:
        """校验用户名不在黑名单且只含合法字符"""
        if v.lower() in cls.FORBIDDEN_USERNAMES:
            raise ValueError(f"用户名 '{v}' 已被系统保留")
        if not v.replace("_", "").replace("-", "").isalnum():
            raise ValueError("用户名只能包含字母、数字、下划线和连字符")
        return v

    @field_validator("email")
    @classmethod
    def email_must_contain_at(cls, v: str) -> str:
        """简单校验邮箱格式"""
        if "@" not in v or "." not in v.split("@")[-1]:
            raise ValueError("邮箱格式不正确")
        return v.strip().lower()

    @field_validator("password_confirm")
    @classmethod
    def passwords_match(cls, v: str, info) -> str:
        """校验两次密码输入一致"""
        # info.data 包含已校验过的字段值
        if "password" in info.data and v != info.data["password"]:
            raise ValueError("两次输入的密码不一致")
        return v


# 使用示例
try:
    user = UserRegistration(
        username="admin",
        password="12345678",
        password_confirm="12345679",
        email="bad-email"
    )
except ValidationError as e:
    for err in e.errors():
        print(f"  {err['loc']}: {err['msg']}")

# Python 3.10+
from pydantic import BaseModel, model_validator
from typing import Self


class DateRange(BaseModel):
    start_date: str
    end_date: str

    @model_validator(mode="after")
    def check_date_order(self) -> Self:
        """确保开始日期不晚于结束日期"""
        if self.start_date > self.end_date:
            raise ValueError(
                f"开始日期 ({self.start_date}) 不能晚于结束日期 ({self.end_date})"
            )
        return self

    @model_validator(mode="before")
    @classmethod
    def normalize_dates(cls, data: dict) -> dict:
        """预处理：如果只传了一个日期，自动填充另一个"""
        if isinstance(data, dict):
            if "start_date" in data and "end_date" not in data:
                data["end_date"] = data["start_date"]
            elif "end_date" in data and "start_date" not in data:
                data["start_date"] = data["end_date"]
        return data


# 使用示例
# 只传一个日期，自动补全
dr = DateRange(start_date="2026-06-01")
print(dr.model_dump())  # {'start_date': '2026-06-01', 'end_date': '2026-06-01'}

# 日期顺序错误
try:
    DateRange(start_date="2026-06-10", end_date="2026-06-01")
except ValidationError as e:
    print(e.errors()[0]["msg"])  # 开始日期不能晚于结束日期

# Python 3.10+
from pydantic import BaseModel
from typing import Optional


class Address(BaseModel):
    street: str
    city: str
    zip_code: str
    country: str = "中国"


class Company(BaseModel):
    name: str
    address: Address  # 嵌套模型
    industry: Optional[str] = None


class Employee(BaseModel):
    id: int
    name: str
    company: Company
    skills: list[str] = []


# 从嵌套 dict 直接实例化
data = {
    "id": 1,
    "name": "张三",
    "company": {
        "name": "TechCorp",
        "address": {
            "street": "科技路 88 号",
            "city": "北京",
            "zip_code": "100000"
        },
        "industry": "互联网"
    },
    "skills": ["Python", "FastAPI", "Pydantic"]
}

emp = Employee(**data)
print(emp.company.address.city)  # 北京
print(emp.model_dump_json(indent=2))

# Python 3.10+
from pydantic import BaseModel
from typing import Optional


class Category(BaseModel):
    name: str
    parent: Optional["Category"] = None  # 自引用（需用字符串形式）
    subcategories: list["Category"] = []


# 构建递归结构
electronics = Category(
    name="电子产品",
    subcategories=[
        Category(
            name="手机",
            subcategories=[
                Category(name="智能手机"),
                Category(name="功能手机"),
            ]
        ),
        Category(name="电脑"),
    ]
)

print(electronics.model_dump_json(indent=2))

# Python 3.10+
from pydantic import BaseModel, ConfigDict


class StrictUser(BaseModel):
    """严格模式配置"""
    model_config = ConfigDict(
        # 禁止额外字段 — 传入未定义的字段会报错
        extra="forbid",
        # 冻结模型 — 实例化后不可修改字段
        frozen=True,
        # 赋值时校验 — 修改属性时触发校验
        validate_assignment=True,
        # 从属性名填充 — 允许用属性名查找别名
        from_attributes=True,
        # 严格类型 — 不自动做类型转换
        strict=True,
    )

    id: int
    name: str
    email: str


# frozen=True 的效果
user = StrictUser(id=1, name="Alice", email="alice@example.com")
try:
    user.name = "Bob"  # 抛出 ValidationError
except ValidationError:
    print("模型已冻结，无法修改字段")

# extra="forbid" 的效果
try:
    StrictUser(id=1, name="Alice", email="alice@example.com", extra_field="oops")
except ValidationError:
    print("不允许传入额外字段")

# Python 3.10+
from pydantic import BaseModel
from typing import Generic, TypeVar

T = TypeVar("T")


class PaginatedResponse(BaseModel, Generic[T]):
    """通用分页响应模型"""
    items: list[T]
    total: int
    page: int
    page_size: int


class UserItem(BaseModel):
    id: int
    name: str


class OrderItem(BaseModel):
    id: str
    amount: float
    status: str


# 使用泛型
user_page = PaginatedResponse[UserItem](
    items=[UserItem(id=1, name="Alice"), UserItem(id=2, name="Bob")],
    total=100,
    page=1,
    page_size=20,
)
print(user_page.model_dump_json(indent=2))

# Python 3.10+
from pydantic import BaseModel
from typing import Annotated, Literal, Union
from typing import TypeAlias


class Cat(BaseModel):
    pet_type: Literal["cat"]
    meows: int                  # 猫叫次数
    favorite_toy: str


class Dog(BaseModel):
    pet_type: Literal["dog"]
    barks: float                # 犬吠音量
    favorite_treat: str


class Fish(BaseModel):
    pet_type: Literal["fish"]
    fins: int                   # 鱼鳍数量
    water_type: str             # 淡水/海水


# 可区分联合 — 根据 pet_type 自动选择模型
Pet: TypeAlias = Annotated[
    Union[Cat, Dog, Fish],
    Field(discriminator="pet_type")
]


class PetOwner(BaseModel):
    name: str
    pet: Pet


# 校验时自动根据 pet_type 选择正确的模型
owner1 = PetOwner(
    name="张三",
    pet={"pet_type": "cat", "meows": 5, "favorite_toy": "毛线球"}
)
print(type(owner1.pet))  # <class '__main__.Cat'>

owner2 = PetOwner(
    name="李四",
    pet={"pet_type": "dog", "barks": 3.5, "favorite_treat": "骨头"}
)
print(type(owner2.pet))  # <class '__main__.Dog'>

# 传入不匹配的字段会报错
try:
    PetOwner(
        name="王五",
        pet={"pet_type": "cat", "barks": 1.0}  # cat 没有 barks 字段
    )
except ValidationError as e:
    print("校验失败：字段不匹配")

# Python 3.10+
from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional


class Article(BaseModel):
    id: int = Field(exclude=True)  # 序列化时排除
    title: str
    content: str
    author: str
    status: str = "draft"
    created_at: datetime = datetime.now()
    secret_note: Optional[str] = Field(default=None, exclude=True)

    # 计算字段（v2 新特性）
    @property
    def summary(self) -> str:
        return self.content[:50] + "..." if len(self.content) > 50 else self.content


article = Article(
    id=123,
    title="Pydantic v2 序列化指南",
    content="Pydantic v2 带来了全新的序列化 API，支持更多控制选项...",
    author="文档组",
    secret_note="内部评审意见"
)

# 基础序列化
print(article.model_dump())
# id 和 secret_note 被 exclude=True 排除

# 指定包含/排除字段
print(article.model_dump(include={"title", "author"}))
# {'title': 'Pydantic v2 序列化指南', 'author': '文档组'}

print(article.model_dump(exclude={"content", "created_at"}))

# 序列化模式
print(article.model_dump(mode="json"))
# datetime 转为 ISO 格式字符串

# 输出 JSON 字符串
json_str = article.model_dump_json(indent=2, exclude_none=True)
print(json_str)

# Python 3.10+
from pydantic import BaseModel
from typing import Optional


class Department(BaseModel):
    name: str
    manager: Optional["Employee"] = None  # 前向引用


class Employee(BaseModel):
    name: str
    department: Department


# 创建实例
dept = Department(name="研发部")
emp = Employee(name="张三", department=dept)
dept.manager = emp  # 设置循环引用

# 序列化时会自动处理循环引用
print(dept.model_dump_json(indent=2))

# Python 3.10+
from pydantic import BaseModel, TypeAdapter
from typing import Annotated


# 1. 避免重复创建模型实例的校验开销
# 使用 TypeAdapter 进行批量校验
class Item(BaseModel):
    name: str
    price: float


items_adapter = TypeAdapter(list[Item])

# 批量校验 — 比逐个实例化快
raw_data = [{"name": f"item_{i}", "price": i * 10.0} for i in range(1000)]
items = items_adapter.validate_python(raw_data)

# 2. 生产环境关闭不必要的校验
# model_config = ConfigDict(revalidate_instances="never")

# 3. 使用 model_construct() 跳过校验（仅在信任数据源时使用）
trusted_data = {"name": "trusted", "price": 99.0}
item = Item.model_construct(**trusted_data)  # 不校验，极快

# Python 3.10+
from pydantic import BaseModel, Field, field_validator, model_validator
from typing import Optional, Self
from enum import Enum
from datetime import date


class OrderStatus(str, Enum):
    PENDING = "pending"
    CONFIRMED = "confirmed"
    SHIPPED = "shipped"
    DELIVERED = "delivered"
    CANCELLED = "cancelled"


class OrderItem(BaseModel):
    product_id: str = Field(min_length=1, max_length=50)
    quantity: int = Field(gt=0, le=9999)
    unit_price: float = Field(gt=0)


class CreateOrderRequest(BaseModel):
    """订单创建请求模型"""
    customer_name: str = Field(min_length=1, max_length=100)
    customer_email: str
    items: list[OrderItem] = Field(min_length=1, max_length=50)
    shipping_address: str = Field(min_length=1, max_length=500)
    coupon_code: Optional[str] = Field(default=None, max_length=20)
    notes: Optional[str] = Field(default=None, max_length=1000)

    @field_validator("customer_email")
    @classmethod
    def validate_email(cls, v: str) -> str:
        if "@" not in v:
            raise ValueError("邮箱格式无效")
        return v.strip().lower()

    @model_validator(mode="after")
    def check_order_total(self) -> Self:
        """校验订单总金额不超过上限"""
        total = sum(item.quantity * item.unit_price for item in self.items)
        if total > 100_000:
            raise ValueError(f"订单总金额 {total:.2f} 超过上限 100,000")
        return self


# 模拟 API 请求
request_data = {
    "customer_name": "张三",
    "customer_email": "zhangsan@example.com",
    "items": [
        {"product_id": "PROD-001", "quantity": 2, "unit_price": 49.99},
        {"product_id": "PROD-002", "quantity": 1, "unit_price": 199.00},
    ],
    "shipping_address": "北京市朝阳区科技路 88 号",
    "coupon_code": "SAVE10",
}

order = CreateOrderRequest(**request_data)
print(f"订单总额: {sum(i.quantity * i.unit_price for i in order.items):.2f}")
# 订单总额: 298.98

# Python 3.10+
# pip install pydantic-settings

from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import Field, field_validator
from typing import Optional


class AppSettings(BaseSettings):
    """应用配置模型"""

    model_config = SettingsConfigDict(
        # 指定 .env 文件路径
        env_file=".env",
        env_file_encoding="utf-8",
        # 环境变量前缀（可选）
        env_prefix="APP_",
        # 是否区分大小写
        case_sensitive=False,
    )

    # 应用基础配置
    app_name: str = "MyApp"
    app_version: str = "1.0.0"
    debug: bool = False

    # 数据库配置
    database_url: str = Field(
        default="sqlite:///./app.db",
        description="数据库连接字符串",
    )
    database_pool_size: int = Field(default=10, ge=1, le=100)

    # Redis 配置
    redis_host: str = "localhost"
    redis_port: int = Field(default=6379, ge=1, le=65535)
    redis_password: Optional[str] = None

    # 安全配置
    secret_key: str = Field(min_length=32)
    access_token_expire_minutes: int = Field(default=30, ge=1)
    allowed_origins: list[str] = Field(default_factory=lambda: ["*"])

    # API 配置
    api_prefix: str = "/api/v1"
    rate_limit_per_minute: int = Field(default=60, ge=1)

    @field_validator("database_url")
    @classmethod
    def check_database_url(cls, v: str) -> str:
        """校验数据库 URL 格式"""
        if not any(v.startswith(prefix) for prefix in
                   ("sqlite://", "postgresql://", "mysql://", "postgresql+asyncpg://")):
            raise ValueError(f"不支持的数据库 URL 格式: {v}")
        return v

    @field_validator("allowed_origins", mode="before")
    @classmethod
    def parse_origins(cls, v: str | list[str]) -> list[str]:
        """支持从环境变量读取逗号分隔的字符串"""
        if isinstance(v, str):
            return [origin.strip() for origin in v.split(",")]
        return v


# 加载配置
settings = AppSettings()  # 自动从环境变量和 .env 文件加载

# 使用配置
print(f"应用: {settings.app_name} v{settings.app_version}")
print(f"数据库: {settings.database_url}")
print(f"Redis: {settings.redis_host}:{settings.redis_port}")
print(f"调试模式: {settings.debug}")

# .env
APP_DEBUG=true
APP_DATABASE_URL=postgresql://user:pass@localhost:5432/mydb
APP_SECRET_KEY=this-is-a-very-secret-key-32-chars
APP_REDIS_HOST=redis-cluster.example.com
APP_REDIS_PORT=6380