pydantic

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Pydantic Data Validation Skill

Pydantic数据验证技能

Overview

概述

You are a Pydantic expert specializing in data validation for customer support systems. Your role is to help build robust, type-safe data models that validate support tickets, user data, API requests, and configuration settings using Pydantic V2.

您是专注于客户支持系统数据验证的Pydantic专家。您的职责是使用Pydantic V2构建健壮、类型安全的数据模型，用于验证支持工单、用户数据、API请求和配置设置。

Core Competencies

核心能力

1. BaseModel Fundamentals

1. BaseModel基础

Purpose: Create validated data models with automatic type coercion and comprehensive error reporting.

Key Principles:

Define models using Python type hints
Leverage automatic validation on instantiation
Use
```
model_dump()
```
and
```
model_dump_json()
```
for serialization
Handle
```
ValidationError
```
exceptions gracefully
Implement proper error logging for support operations

Basic Pattern:

python

from pydantic import BaseModel, Field, ValidationError
from datetime import datetime
from typing import Optional

class SupportTicket(BaseModel):
    ticket_id: int
    customer_email: str
    subject: str = Field(min_length=5, max_length=200)
    description: str = Field(min_length=20)
    priority: str = Field(pattern=r'^(low|medium|high|urgent)$')
    created_at: datetime
    assigned_to: Optional[str] = None
    status: str = 'open'

try:
    ticket = SupportTicket(
        ticket_id=12345,
        customer_email='customer@example.com',
        subject='Login Issue',
        description='Cannot access my account after password reset',
        priority='high',
        created_at='2024-01-15T10:30:00'
    )
    print(ticket.model_dump())
except ValidationError as e:
    # Log validation errors for support team review
    for error in e.errors():
        print(f"Field: {error['loc']}, Error: {error['msg']}")

用途：创建具备自动类型转换和全面错误报告的验证数据模型。

核心原则:

使用Python类型提示定义模型
利用实例化时的自动验证
使用
```
model_dump()
```
和
```
model_dump_json()
```
进行序列化
优雅处理
```
ValidationError
```
异常
为支持操作实现合理的错误日志

基础模式:

python

from pydantic import BaseModel, Field, ValidationError
from datetime import datetime
from typing import Optional

class SupportTicket(BaseModel):
    ticket_id: int
    customer_email: str
    subject: str = Field(min_length=5, max_length=200)
    description: str = Field(min_length=20)
    priority: str = Field(pattern=r'^(low|medium|high|urgent)$')
    created_at: datetime
    assigned_to: Optional[str] = None
    status: str = 'open'

try:
    ticket = SupportTicket(
        ticket_id=12345,
        customer_email='customer@example.com',
        subject='Login Issue',
        description='Cannot access my account after password reset',
        priority='high',
        created_at='2024-01-15T10:30:00'
    )
    print(ticket.model_dump())
except ValidationError as e:
    # Log validation errors for support team review
    for error in e.errors():
        print(f"Field: {error['loc']}, Error: {error['msg']}")

2. Field Configuration and Constraints

2. 字段配置与约束

Purpose: Apply granular validation rules to individual fields for data quality assurance.

Common Constraints:

String validation:

min_length

max_length

pattern

strip_whitespace

Numeric validation:
```
gt
```
,
```
ge
```
,
```
lt
```
,
```
le
```
,
```
multiple_of
```

Field metadata:

title

description

examples

json_schema_extra

Serialization control:
```
alias
```
,
```
serialization_alias
```
,
```
exclude
```
,
```
include
```

Customer Support Example:

python

from pydantic import BaseModel, Field, EmailStr, HttpUrl
from typing import Annotated
from datetime import datetime

class CustomerProfile(BaseModel):
    # ID fields with constraints
    customer_id: Annotated[int, Field(gt=0, description="Unique customer identifier")]

    # Contact information with validation
    email: EmailStr
    phone: Annotated[str, Field(pattern=r'^\+?1?\d{9,15}$', description="International phone format")]

    # Name fields with length constraints
    first_name: Annotated[str, Field(min_length=1, max_length=50, strip_whitespace=True)]
    last_name: Annotated[str, Field(min_length=1, max_length=50, strip_whitespace=True)]

    # Company information (optional)
    company_name: Optional[Annotated[str, Field(max_length=100)]] = None
    company_website: Optional[HttpUrl] = None

    # Support tier with default
    support_tier: Annotated[str, Field(pattern=r'^(basic|premium|enterprise)$')] = 'basic'

    # Metadata fields
    registration_date: datetime
    last_contact: Optional[datetime] = None
    notes: str = Field(default='', max_length=2000, description="Internal notes")

    # Excluded from serialization (internal use only)
    internal_score: int = Field(default=0, exclude=True)

    model_config = {
        'str_strip_whitespace': True,
        'validate_assignment': True,
        'populate_by_name': True
    }

用途：为单个字段应用细粒度的验证规则，确保数据质量。

常见约束:

字符串验证：

min_length

max_length

pattern

strip_whitespace

数值验证：
```
gt
```
,
```
ge
```
,
```
lt
```
,
```
le
```
,
```
multiple_of
```

字段元数据：

title

description

examples

json_schema_extra

序列化控制：
```
alias
```
,
```
serialization_alias
```
,
```
exclude
```
,
```
include
```

客户支持示例:

python

from pydantic import BaseModel, Field, EmailStr, HttpUrl
from typing import Annotated
from datetime import datetime

class CustomerProfile(BaseModel):
    # ID字段带约束
    customer_id: Annotated[int, Field(gt=0, description="Unique customer identifier")]

    # 带验证的联系信息
    email: EmailStr
    phone: Annotated[str, Field(pattern=r'^\+?1?\d{9,15}$', description="International phone format")]

    # 带长度约束的姓名字段
    first_name: Annotated[str, Field(min_length=1, max_length=50, strip_whitespace=True)]
    last_name: Annotated[str, Field(min_length=1, max_length=50, strip_whitespace=True)]

    # 公司信息（可选）
    company_name: Optional[Annotated[str, Field(max_length=100)]] = None
    company_website: Optional[HttpUrl] = None

    # 带默认值的支持层级
    support_tier: Annotated[str, Field(pattern=r'^(basic|premium|enterprise)$')] = 'basic'

    # 元数据字段
    registration_date: datetime
    last_contact: Optional[datetime] = None
    notes: str = Field(default='', max_length=2000, description="Internal notes")

    # 序列化时排除（仅内部使用）
    internal_score: int = Field(default=0, exclude=True)

    model_config = {
        'str_strip_whitespace': True,
        'validate_assignment': True,
        'populate_by_name': True
    }

3. Custom Field Validators

3. 自定义字段验证器

Purpose: Implement business logic validation beyond basic type checking.

@field_validator Pattern:

python

from pydantic import BaseModel, field_validator, ValidationInfo
import re

class SupportTicketSubmission(BaseModel):
    customer_email: str
    subject: str
    description: str
    category: str
    attachments: list[str] = []

    @field_validator('customer_email')
    @classmethod
    def validate_email_domain(cls, v: str) -> str:
        """Validate email format and check against blocked domains"""
        blocked_domains = ['tempmail.com', 'throwaway.email']

        if '@' not in v:
            raise ValueError('Invalid email format')

        domain = v.split('@')[1].lower()
        if domain in blocked_domains:
            raise ValueError(f'Email domain {domain} is not allowed')

        return v.lower()

    @field_validator('subject')
    @classmethod
    def validate_subject(cls, v: str) -> str:
        """Ensure subject is meaningful and not spam"""
        v = v.strip()

        # Check for minimum word count
        words = v.split()
        if len(words) < 2:
            raise ValueError('Subject must contain at least 2 words')

        # Check for spam patterns
        spam_patterns = [r'viagra', r'casino', r'lottery']
        for pattern in spam_patterns:
            if re.search(pattern, v, re.IGNORECASE):
                raise ValueError('Subject contains prohibited content')

        return v

    @field_validator('attachments')
    @classmethod
    def validate_attachments(cls, v: list[str]) -> list[str]:
        """Validate attachment file extensions"""
        allowed_extensions = {'.pdf', '.jpg', '.jpeg', '.png', '.doc', '.docx', '.txt'}

        for filename in v:
            ext = filename[filename.rfind('.'):].lower() if '.' in filename else ''
            if ext not in allowed_extensions:
                raise ValueError(f'File type {ext} not allowed. Allowed: {allowed_extensions}')

        if len(v) > 5:
            raise ValueError('Maximum 5 attachments allowed')

        return v

    @field_validator('category')
    @classmethod
    def validate_category(cls, v: str) -> str:
        """Normalize and validate ticket category"""
        valid_categories = {
            'technical', 'billing', 'account', 'feature_request',
            'bug_report', 'general_inquiry'
        }

        v_normalized = v.lower().replace(' ', '_')
        if v_normalized not in valid_categories:
            raise ValueError(f'Invalid category. Valid options: {valid_categories}')

        return v_normalized

用途：实现基础类型检查之外的业务逻辑验证。

@field_validator模式:

python

from pydantic import BaseModel, field_validator, ValidationInfo
import re

class SupportTicketSubmission(BaseModel):
    customer_email: str
    subject: str
    description: str
    category: str
    attachments: list[str] = []

    @field_validator('customer_email')
    @classmethod
    def validate_email_domain(cls, v: str) -> str:
        """Validate email format and check against blocked domains"""
        blocked_domains = ['tempmail.com', 'throwaway.email']

        if '@' not in v:
            raise ValueError('Invalid email format')

        domain = v.split('@')[1].lower()
        if domain in blocked_domains:
            raise ValueError(f'Email domain {domain} is not allowed')

        return v.lower()

    @field_validator('subject')
    @classmethod
    def validate_subject(cls, v: str) -> str:
        """Ensure subject is meaningful and not spam"""
        v = v.strip()

        # 检查最小单词数
        words = v.split()
        if len(words) < 2:
            raise ValueError('Subject must contain at least 2 words')

        # 检查垃圾内容模式
        spam_patterns = [r'viagra', r'casino', r'lottery']
        for pattern in spam_patterns:
            if re.search(pattern, v, re.IGNORECASE):
                raise ValueError('Subject contains prohibited content')

        return v

    @field_validator('attachments')
    @classmethod
    def validate_attachments(cls, v: list[str]) -> list[str]:
        """Validate attachment file extensions"""
        allowed_extensions = {'.pdf', '.jpg', '.jpeg', '.png', '.doc', '.docx', '.txt'}

        for filename in v:
            ext = filename[filename.rfind('.'):].lower() if '.' in filename else ''
            if ext not in allowed_extensions:
                raise ValueError(f'File type {ext} not allowed. Allowed: {allowed_extensions}')

        if len(v) > 5:
            raise ValueError('Maximum 5 attachments allowed')

        return v

    @field_validator('category')
    @classmethod
    def validate_category(cls, v: str) -> str:
        """Normalize and validate ticket category"""
        valid_categories = {
            'technical', 'billing', 'account', 'feature_request',
            'bug_report', 'general_inquiry'
        }

        v_normalized = v.lower().replace(' ', '_')
        if v_normalized not in valid_categories:
            raise ValueError(f'Invalid category. Valid options: {valid_categories}')

        return v_normalized

4. Model-Level Validation

4. 模型级验证

Purpose: Validate relationships between multiple fields and perform cross-field validation.

@model_validator Pattern:

python

from pydantic import BaseModel, model_validator, ValidationError
from datetime import datetime, timedelta
from typing import Any, Optional

class TicketSchedule(BaseModel):
    ticket_id: int
    scheduled_start: datetime
    scheduled_end: datetime
    technician_id: Optional[int] = None
    estimated_hours: float
    priority: str

    @model_validator(mode='before')
    @classmethod
    def preprocess_data(cls, data: Any) -> Any:
        """Preprocess and normalize data before field validation"""
        if isinstance(data, dict):
            # Auto-generate estimated hours if not provided
            if 'scheduled_start' in data and 'scheduled_end' in data and 'estimated_hours' not in data:
                start = datetime.fromisoformat(data['scheduled_start'])
                end = datetime.fromisoformat(data['scheduled_end'])
                data['estimated_hours'] = (end - start).total_seconds() / 3600

            # Normalize priority
            if 'priority' in data:
                data['priority'] = data['priority'].lower()

        return data

    @model_validator(mode='after')
    def validate_schedule(self) -> 'TicketSchedule':
        """Validate scheduling logic after all fields are validated"""
        # Ensure end is after start
        if self.scheduled_end <= self.scheduled_start:
            raise ValueError('scheduled_end must be after scheduled_start')

        # Validate duration against priority
        duration = (self.scheduled_end - self.scheduled_start).total_seconds() / 3600

        if self.priority == 'urgent' and duration > 2:
            raise ValueError('Urgent tickets must be scheduled within 2 hours')

        if self.priority == 'low' and duration < 1:
            raise ValueError('Low priority tickets require minimum 1 hour allocation')

        # Ensure estimated hours match duration
        if abs(self.estimated_hours - duration) > 0.1:
            raise ValueError('estimated_hours must match scheduled duration')

        # Validate business hours (9 AM - 6 PM)
        if self.scheduled_start.hour < 9 or self.scheduled_start.hour >= 18:
            raise ValueError('Tickets must be scheduled during business hours (9 AM - 6 PM)')

        return self

class TicketEscalation(BaseModel):
    ticket_id: int
    current_assignee: str
    escalation_level: int
    reason: str
    requested_by: str
    approved_by: Optional[str] = None

    @model_validator(mode='after')
    def validate_escalation_approval(self) -> 'TicketEscalation':
        """Ensure high-level escalations require approval"""
        if self.escalation_level >= 3 and not self.approved_by:
            raise ValueError('Level 3+ escalations require manager approval')

        if self.approved_by == self.requested_by:
            raise ValueError('Approver must be different from requester')

        return self

用途：验证多个字段之间的关系，执行跨字段验证。

@model_validator模式:

python

from pydantic import BaseModel, model_validator, ValidationError
from datetime import datetime, timedelta
from typing import Any, Optional

class TicketSchedule(BaseModel):
    ticket_id: int
    scheduled_start: datetime
    scheduled_end: datetime
    technician_id: Optional[int] = None
    estimated_hours: float
    priority: str

    @model_validator(mode='before')
    @classmethod
    def preprocess_data(cls, data: Any) -> Any:
        """Preprocess and normalize data before field validation"""
        if isinstance(data, dict):
            # 若未提供则自动计算预估时长
            if 'scheduled_start' in data and 'scheduled_end' in data and 'estimated_hours' not in data:
                start = datetime.fromisoformat(data['scheduled_start'])
                end = datetime.fromisoformat(data['scheduled_end'])
                data['estimated_hours'] = (end - start).total_seconds() / 3600

            # 标准化优先级
            if 'priority' in data:
                data['priority'] = data['priority'].lower()

        return data

    @model_validator(mode='after')
    def validate_schedule(self) -> 'TicketSchedule':
        """Validate scheduling logic after all fields are validated"""
        # 确保结束时间晚于开始时间
        if self.scheduled_end <= self.scheduled_start:
            raise ValueError('scheduled_end must be after scheduled_start')

        # 根据优先级验证时长
        duration = (self.scheduled_end - self.scheduled_start).total_seconds() / 3600

        if self.priority == 'urgent' and duration > 2:
            raise ValueError('Urgent tickets must be scheduled within 2 hours')

        if self.priority == 'low' and duration < 1:
            raise ValueError('Low priority tickets require minimum 1 hour allocation')

        # 确保预估时长与实际时长匹配
        if abs(self.estimated_hours - duration) > 0.1:
            raise ValueError('estimated_hours must match scheduled duration')

        # 验证工作时间（上午9点 - 下午6点）
        if self.scheduled_start.hour < 9 or self.scheduled_start.hour >= 18:
            raise ValueError('Tickets must be scheduled during business hours (9 AM - 6 PM)')

        return self

class TicketEscalation(BaseModel):
    ticket_id: int
    current_assignee: str
    escalation_level: int
    reason: str
    requested_by: str
    approved_by: Optional[str] = None

    @model_validator(mode='after')
    def validate_escalation_approval(self) -> 'TicketEscalation':
        """Ensure high-level escalations require approval"""
        if self.escalation_level >= 3 and not self.approved_by:
            raise ValueError('Level 3+ escalations require manager approval')

        if self.approved_by == self.requested_by:
            raise ValueError('Approver must be different from requester')

        return self

5. Nested Models and Complex Structures

5. 嵌套模型与复杂结构

Purpose: Build hierarchical data models for complex support workflows.

Pattern:

python

from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional, Literal
from enum import Enum

class TicketStatus(str, Enum):
    OPEN = 'open'
    IN_PROGRESS = 'in_progress'
    PENDING_CUSTOMER = 'pending_customer'
    RESOLVED = 'resolved'
    CLOSED = 'closed'

class Comment(BaseModel):
    comment_id: int
    author: str
    content: str = Field(min_length=1, max_length=5000)
    timestamp: datetime
    is_internal: bool = False

class Attachment(BaseModel):
    filename: str
    file_size_bytes: int = Field(gt=0, le=10_000_000)  # Max 10MB
    content_type: str
    uploaded_by: str
    uploaded_at: datetime
    url: str

class Resolution(BaseModel):
    resolved_by: str
    resolution_date: datetime
    resolution_summary: str = Field(min_length=20, max_length=2000)
    root_cause: Optional[str] = None
    preventive_measures: Optional[str] = None
    customer_satisfaction_score: Optional[int] = Field(default=None, ge=1, le=5)

class CustomerInfo(BaseModel):
    customer_id: int
    name: str
    email: str
    phone: Optional[str] = None
    account_type: Literal['free', 'basic', 'premium', 'enterprise']
    registration_date: datetime

class CompleteTicket(BaseModel):
    # Core ticket information
    ticket_id: int
    customer: CustomerInfo

    # Ticket details
    subject: str = Field(min_length=5, max_length=200)
    description: str = Field(min_length=20)
    category: str
    priority: str
    status: TicketStatus

    # Assignment and tracking
    assigned_to: Optional[str] = None
    created_at: datetime
    updated_at: datetime

    # Rich content
    comments: list[Comment] = []
    attachments: list[Attachment] = []

    # Resolution (if resolved/closed)
    resolution: Optional[Resolution] = None

    # Metadata
    tags: list[str] = Field(default_factory=list, max_length=10)
    related_tickets: list[int] = Field(default_factory=list)

    @model_validator(mode='after')
    def validate_ticket_state(self) -> 'CompleteTicket':
        """Ensure ticket state is consistent"""
        if self.status in (TicketStatus.RESOLVED, TicketStatus.CLOSED) and not self.resolution:
            raise ValueError('Resolved/closed tickets must have resolution details')

        if self.status == TicketStatus.IN_PROGRESS and not self.assigned_to:
            raise ValueError('In-progress tickets must be assigned')

        return self

    model_config = {
        'use_enum_values': True,
        'validate_assignment': True,
        'json_schema_extra': {
            'examples': [{
                'ticket_id': 12345,
                'customer': {
                    'customer_id': 6789,
                    'name': 'John Doe',
                    'email': 'john@example.com',
                    'account_type': 'premium',
                    'registration_date': '2023-01-01T00:00:00'
                },
                'subject': 'Unable to access dashboard',
                'description': 'Getting 403 error when trying to access analytics dashboard',
                'category': 'technical',
                'priority': 'high',
                'status': 'in_progress',
                'assigned_to': 'tech_support_1',
                'created_at': '2024-01-15T10:00:00',
                'updated_at': '2024-01-15T11:30:00'
            }]
        }
    }

用途：为复杂的支持工作流构建分层数据模型。

模式:

python

from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional, Literal
from enum import Enum

class TicketStatus(str, Enum):
    OPEN = 'open'
    IN_PROGRESS = 'in_progress'
    PENDING_CUSTOMER = 'pending_customer'
    RESOLVED = 'resolved'
    CLOSED = 'closed'

class Comment(BaseModel):
    comment_id: int
    author: str
    content: str = Field(min_length=1, max_length=5000)
    timestamp: datetime
    is_internal: bool = False

class Attachment(BaseModel):
    filename: str
    file_size_bytes: int = Field(gt=0, le=10_000_000)  # 最大10MB
    content_type: str
    uploaded_by: str
    uploaded_at: datetime
    url: str

class Resolution(BaseModel):
    resolved_by: str
    resolution_date: datetime
    resolution_summary: str = Field(min_length=20, max_length=2000)
    root_cause: Optional[str] = None
    preventive_measures: Optional[str] = None
    customer_satisfaction_score: Optional[int] = Field(default=None, ge=1, le=5)

class CustomerInfo(BaseModel):
    customer_id: int
    name: str
    email: str
    phone: Optional[str] = None
    account_type: Literal['free', 'basic', 'premium', 'enterprise']
    registration_date: datetime

class CompleteTicket(BaseModel):
    # 核心工单信息
    ticket_id: int
    customer: CustomerInfo

    # 工单详情
    subject: str = Field(min_length=5, max_length=200)
    description: str = Field(min_length=20)
    category: str
    priority: str
    status: TicketStatus

    # 分配与跟踪
    assigned_to: Optional[str] = None
    created_at: datetime
    updated_at: datetime

    # 富内容
    comments: list[Comment] = []
    attachments: list[Attachment] = []

    # 解决信息（若工单已解决/关闭）
    resolution: Optional[Resolution] = None

    # 元数据
    tags: list[str] = Field(default_factory=list, max_length=10)
    related_tickets: list[int] = Field(default_factory=list)

    @model_validator(mode='after')
    def validate_ticket_state(self) -> 'CompleteTicket':
        """Ensure ticket state is consistent"""
        if self.status in (TicketStatus.RESOLVED, TicketStatus.CLOSED) and not self.resolution:
            raise ValueError('Resolved/closed tickets must have resolution details')

        if self.status == TicketStatus.IN_PROGRESS and not self.assigned_to:
            raise ValueError('In-progress tickets must be assigned')

        return self

    model_config = {
        'use_enum_values': True,
        'validate_assignment': True,
        'json_schema_extra': {
            'examples': [{
                'ticket_id': 12345,
                'customer': {
                    'customer_id': 6789,
                    'name': 'John Doe',
                    'email': 'john@example.com',
                    'account_type': 'premium',
                    'registration_date': '2023-01-01T00:00:00'
                },
                'subject': 'Unable to access dashboard',
                'description': 'Getting 403 error when trying to access analytics dashboard',
                'category': 'technical',
                'priority': 'high',
                'status': 'in_progress',
                'assigned_to': 'tech_support_1',
                'created_at': '2024-01-15T10:00:00',
                'updated_at': '2024-01-15T11:30:00'
            }]
        }
    }

6. FastAPI Integration for Request/Response Models

6. FastAPI集成：请求/响应模型

Purpose: Create type-safe API endpoints with automatic validation and documentation.

Pattern:

python

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field, ValidationError
from typing import Optional, List
from datetime import datetime

app = FastAPI(title="Support Ticket API")

用途：创建具备自动验证和文档的类型安全API端点。

模式:

python

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field, ValidationError
from typing import Optional, List
from datetime import datetime

app = FastAPI(title="Support Ticket API")

Request Models

请求模型

class TicketCreateRequest(BaseModel): customer_email: str subject: str = Field(min_length=5, max_length=200) description: str = Field(min_length=20, max_length=5000) category: str priority: str = Field(default='medium', pattern=r'^(low|medium|high|urgent)$') attachments: List[str] = Field(default_factory=list, max_length=5)

model_config = {
    'json_schema_extra': {
        'examples': [{
            'customer_email': 'user@example.com',
            'subject': 'Cannot reset password',
            'description': 'I clicked the reset password link but did not receive an email',
            'category': 'account',
            'priority': 'high',
            'attachments': []
        }]
    }
}

class TicketUpdateRequest(BaseModel): subject: Optional[str] = Field(default=None, min_length=5, max_length=200) description: Optional[str] = Field(default=None, min_length=20) status: Optional[str] = None assigned_to: Optional[str] = None priority: Optional[str] = Field(default=None, pattern=r'^(low|medium|high|urgent)$')

model_config = {
    'json_schema_extra': {
        'examples': [{
            'customer_email': 'user@example.com',
            'subject': 'Cannot reset password',
            'description': 'I clicked the reset password link but did not receive an email',
            'category': 'account',
            'priority': 'high',
            'attachments': []
        }]
    }
}

Response Models

响应模型

class TicketResponse(BaseModel): ticket_id: int customer_email: str subject: str description: str category: str priority: str status: str assigned_to: Optional[str] created_at: datetime updated_at: datetime

model_config = {'from_attributes': True}  # For ORM compatibility

class TicketListResponse(BaseModel): tickets: List[TicketResponse] total: int page: int page_size: int

class ErrorResponse(BaseModel): error_code: str message: str details: Optional[dict] = None

model_config = {'from_attributes': True}  # 兼容ORM

class TicketListResponse(BaseModel): tickets: List[TicketResponse] total: int page: int page_size: int

class ErrorResponse(BaseModel): error_code: str message: str details: Optional[dict] = None

API Endpoints

API端点

@app.post( "/tickets/", response_model=TicketResponse, status_code=status.HTTP_201_CREATED, responses={ 400: {"model": ErrorResponse, "description": "Validation error"}, 500: {"model": ErrorResponse, "description": "Server error"} } ) async def create_ticket(ticket: TicketCreateRequest): """Create a new support ticket with automatic validation""" try: # Business logic here # ticket_data is automatically validated by Pydantic new_ticket = { "ticket_id": 12345, "customer_email": ticket.customer_email, "subject": ticket.subject, "description": ticket.description, "category": ticket.category, "priority": ticket.priority, "status": "open", "assigned_to": None, "created_at": datetime.now(), "updated_at": datetime.now() } return TicketResponse(**new_ticket) except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail={"error_code": "CREATION_FAILED", "message": str(e)} )

@app.get("/tickets/", response_model=TicketListResponse) async def list_tickets( page: int = Field(default=1, ge=1), page_size: int = Field(default=20, ge=1, le=100), status: Optional[str] = None, priority: Optional[str] = None ): """List tickets with pagination and filtering""" # Query logic here tickets = [] # Fetch from database return TicketListResponse( tickets=tickets, total=0, page=page, page_size=page_size )

@app.patch("/tickets/{ticket_id}", response_model=TicketResponse) async def update_ticket(ticket_id: int, update: TicketUpdateRequest): """Update ticket with partial validation""" # Only provided fields are validated update_data = update.model_dump(exclude_unset=True) # Apply updates to database # Return updated ticket pass

undefined

@app.post( "/tickets/", response_model=TicketResponse, status_code=status.HTTP_201_CREATED, responses={ 400: {"model": ErrorResponse, "description": "Validation error"}, 500: {"model": ErrorResponse, "description": "Server error"} } ) async def create_ticket(ticket: TicketCreateRequest): """Create a new support ticket with automatic validation""" try: # 业务逻辑在此处 # ticket_data已由Pydantic自动验证 new_ticket = { "ticket_id": 12345, "customer_email": ticket.customer_email, "subject": ticket.subject, "description": ticket.description, "category": ticket.category, "priority": ticket.priority, "status": "open", "assigned_to": None, "created_at": datetime.now(), "updated_at": datetime.now() } return TicketResponse(**new_ticket) except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail={"error_code": "CREATION_FAILED", "message": str(e)} )

@app.get("/tickets/", response_model=TicketListResponse) async def list_tickets( page: int = Field(default=1, ge=1), page_size: int = Field(default=20, ge=1, le=100), status: Optional[str] = None, priority: Optional[str] = None ): """List tickets with pagination and filtering""" # 查询逻辑在此处 tickets = [] # 从数据库获取 return TicketListResponse( tickets=tickets, total=0, page=page, page_size=page_size )

@app.patch("/tickets/{ticket_id}", response_model=TicketResponse) async def update_ticket(ticket_id: int, update: TicketUpdateRequest): """Update ticket with partial validation""" # 仅验证提供的字段 update_data = update.model_dump(exclude_unset=True) # 将更新应用到数据库 # 返回更新后的工单 pass

undefined

7. BaseSettings for Configuration Management

7. BaseSettings：配置管理

Purpose: Manage application configuration from environment variables with validation.

Pattern:

python

from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import Field, PostgresDsn, RedisDsn, EmailStr
from typing import Optional, List

class DatabaseSettings(BaseSettings):
    """Database configuration"""
    host: str = Field(default='localhost', description='Database host')
    port: int = Field(default=5432, ge=1, le=65535)
    username: str
    password: str
    database: str = Field(default='support_db')

    # Computed property for connection string
    @property
    def connection_url(self) -> str:
        return f"postgresql://{self.username}:{self.password}@{self.host}:{self.port}/{self.database}"

    model_config = SettingsConfigDict(
        env_prefix='DB_',
        env_file='.env',
        env_file_encoding='utf-8'
    )

class RedisSettings(BaseSettings):
    """Redis cache configuration"""
    url: RedisDsn = Field(default='redis://localhost:6379/0')
    max_connections: int = Field(default=50, ge=1)
    socket_timeout: int = Field(default=5, ge=1)

    model_config = SettingsConfigDict(
        env_prefix='REDIS_',
        env_file='.env'
    )

class EmailSettings(BaseSettings):
    """Email notification configuration"""
    smtp_host: str = Field(default='smtp.gmail.com')
    smtp_port: int = Field(default=587, ge=1, le=65535)
    smtp_username: str
    smtp_password: str
    from_email: EmailStr
    support_email: EmailStr
    use_tls: bool = Field(default=True)

    model_config = SettingsConfigDict(
        env_prefix='EMAIL_',
        env_file='.env'
    )

class SupportSettings(BaseSettings):
    """Support system configuration"""
    max_tickets_per_page: int = Field(default=50, ge=1, le=200)
    max_attachment_size_mb: int = Field(default=10, ge=1, le=100)
    allowed_file_extensions: List[str] = Field(
        default=['.pdf', '.jpg', '.jpeg', '.png', '.doc', '.docx']
    )
    auto_assign_enabled: bool = Field(default=True)
    escalation_hours: int = Field(default=24, ge=1)
    business_hours_start: int = Field(default=9, ge=0, le=23)
    business_hours_end: int = Field(default=18, ge=0, le=23)

    model_config = SettingsConfigDict(
        env_prefix='SUPPORT_',
        env_file='.env'
    )

class ApplicationSettings(BaseSettings):
    """Main application settings"""
    app_name: str = Field(default='Support Ticket System')
    app_version: str = Field(default='1.0.0')
    debug: bool = Field(default=False)
    api_key: str
    secret_key: str
    allowed_origins: List[str] = Field(default=['http://localhost:3000'])

    # Nested settings
    database: DatabaseSettings = Field(default_factory=DatabaseSettings)
    redis: RedisSettings = Field(default_factory=RedisSettings)
    email: EmailSettings = Field(default_factory=EmailSettings)
    support: SupportSettings = Field(default_factory=SupportSettings)

    model_config = SettingsConfigDict(
        env_file='.env',
        env_file_encoding='utf-8',
        case_sensitive=False,
        extra='ignore'
    )

用途：通过环境变量管理应用配置并进行验证。

模式:

python

from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import Field, PostgresDsn, RedisDsn, EmailStr
from typing import Optional, List

class DatabaseSettings(BaseSettings):
    """Database configuration"""
    host: str = Field(default='localhost', description='Database host')
    port: int = Field(default=5432, ge=1, le=65535)
    username: str
    password: str
    database: str = Field(default='support_db')

    # 计算属性：连接字符串
    @property
    def connection_url(self) -> str:
        return f"postgresql://{self.username}:{self.password}@{self.host}:{self.port}/{self.database}"

    model_config = SettingsConfigDict(
        env_prefix='DB_',
        env_file='.env',
        env_file_encoding='utf-8'
    )

class RedisSettings(BaseSettings):
    """Redis cache configuration"""
    url: RedisDsn = Field(default='redis://localhost:6379/0')
    max_connections: int = Field(default=50, ge=1)
    socket_timeout: int = Field(default=5, ge=1)

    model_config = SettingsConfigDict(
        env_prefix='REDIS_',
        env_file='.env'
    )

class EmailSettings(BaseSettings):
    """Email notification configuration"""
    smtp_host: str = Field(default='smtp.gmail.com')
    smtp_port: int = Field(default=587, ge=1, le=65535)
    smtp_username: str
    smtp_password: str
    from_email: EmailStr
    support_email: EmailStr
    use_tls: bool = Field(default=True)

    model_config = SettingsConfigDict(
        env_prefix='EMAIL_',
        env_file='.env'
    )

class SupportSettings(BaseSettings):
    """Support system configuration"""
    max_tickets_per_page: int = Field(default=50, ge=1, le=200)
    max_attachment_size_mb: int = Field(default=10, ge=1, le=100)
    allowed_file_extensions: List[str] = Field(
        default=['.pdf', '.jpg', '.jpeg', '.png', '.doc', '.docx']
    )
    auto_assign_enabled: bool = Field(default=True)
    escalation_hours: int = Field(default=24, ge=1)
    business_hours_start: int = Field(default=9, ge=0, le=23)
    business_hours_end: int = Field(default=18, ge=0, le=23)

    model_config = SettingsConfigDict(
        env_prefix='SUPPORT_',
        env_file='.env'
    )

class ApplicationSettings(BaseSettings):
    """Main application settings"""
    app_name: str = Field(default='Support Ticket System')
    app_version: str = Field(default='1.0.0')
    debug: bool = Field(default=False)
    api_key: str
    secret_key: str
    allowed_origins: List[str] = Field(default=['http://localhost:3000'])

    # 嵌套配置
    database: DatabaseSettings = Field(default_factory=DatabaseSettings)
    redis: RedisSettings = Field(default_factory=RedisSettings)
    email: EmailSettings = Field(default_factory=EmailSettings)
    support: SupportSettings = Field(default_factory=SupportSettings)

    model_config = SettingsConfigDict(
        env_file='.env',
        env_file_encoding='utf-8',
        case_sensitive=False,
        extra='ignore'
    )

Usage

使用示例

settings = ApplicationSettings() print(f"Connecting to database at: {settings.database.connection_url}") print(f"Max tickets per page: {settings.support.max_tickets_per_page}")

undefined

settings = ApplicationSettings() print(f"Connecting to database at: {settings.database.connection_url}") print(f"Max tickets per page: {settings.support.max_tickets_per_page}")

undefined

8. Serialization and Deserialization

8. 序列化与反序列化

Purpose: Control how models are converted to/from dictionaries, JSON, and other formats.

Pattern:

python

from pydantic import BaseModel, Field, field_serializer, computed_field
from datetime import datetime
from typing import Optional

class TicketExport(BaseModel):
    ticket_id: int
    customer_email: str
    subject: str
    created_at: datetime
    status: str
    priority: str
    assigned_to: Optional[str] = None
    internal_notes: str = Field(default='', exclude=True)

    @field_serializer('customer_email')
    def mask_email(self, email: str) -> str:
        """Mask email for privacy in exports"""
        if '@' in email:
            local, domain = email.split('@')
            masked = local[:2] + '***' + local[-1:] if len(local) > 3 else '***'
            return f"{masked}@{domain}"
        return email

    @field_serializer('created_at')
    def format_datetime(self, dt: datetime) -> str:
        """Format datetime for export"""
        return dt.strftime('%Y-%m-%d %H:%M:%S')

    @computed_field
    @property
    def days_open(self) -> int:
        """Calculate days since ticket creation"""
        return (datetime.now() - self.created_at).days

用途：控制模型如何转换为/从字典、JSON及其他格式。

模式:

python

from pydantic import BaseModel, Field, field_serializer, computed_field
from datetime import datetime
from typing import Optional

class TicketExport(BaseModel):
    ticket_id: int
    customer_email: str
    subject: str
    created_at: datetime
    status: str
    priority: str
    assigned_to: Optional[str] = None
    internal_notes: str = Field(default='', exclude=True)

    @field_serializer('customer_email')
    def mask_email(self, email: str) -> str:
        """Mask email for privacy in exports"""
        if '@' in email:
            local, domain = email.split('@')
            masked = local[:2] + '***' + local[-1:] if len(local) > 3 else '***'
            return f"{masked}@{domain}"
        return email

    @field_serializer('created_at')
    def format_datetime(self, dt: datetime) -> str:
        """Format datetime for export"""
        return dt.strftime('%Y-%m-%d %H:%M:%S')

    @computed_field
    @property
    def days_open(self) -> int:
        """Calculate days since ticket creation"""
        return (datetime.now() - self.created_at).days

Serialization modes

序列化模式

ticket = TicketExport( ticket_id=123, customer_email='john.doe@example.com', subject='Login issue', created_at=datetime.now(), status='open', priority='high', internal_notes='Customer called twice' )

Standard serialization

标准序列化

print(ticket.model_dump())

{'ticket_id': 123, 'customer_email': 'jo***e@example.com', ...}

Include all fields (even excluded)

包含所有字段（即使是被排除的）

print(ticket.model_dump(mode='python', exclude_none=False))

Serialize to JSON

序列化为JSON

json_str = ticket.model_dump_json(indent=2) print(json_str)

Exclude specific fields

排除特定字段

print(ticket.model_dump(exclude={'internal_notes', 'assigned_to'}))

Include only specific fields

仅包含特定字段

print(ticket.model_dump(include={'ticket_id', 'subject', 'status'}))

undefined

print(ticket.model_dump(include={'ticket_id', 'subject', 'status'}))

undefined

9. Advanced Validation Techniques

9. 高级验证技术

Purpose: Implement sophisticated validation logic for complex business requirements.

Pattern:

python

from pydantic import BaseModel, field_validator, model_validator
from typing import Any, Optional
import re

class TicketPrioritization(BaseModel):
    customer_tier: str
    issue_category: str
    response_time_hours: int
    business_impact: str
    affected_users: int = Field(ge=1)

    @field_validator('customer_tier')
    @classmethod
    def validate_tier(cls, v: str) -> str:
        valid_tiers = {'free', 'basic', 'premium', 'enterprise'}
        v = v.lower()
        if v not in valid_tiers:
            raise ValueError(f'Invalid tier. Must be one of: {valid_tiers}')
        return v

    @model_validator(mode='after')
    def calculate_priority(self) -> 'TicketPrioritization':
        """Auto-calculate priority based on multiple factors"""
        priority_score = 0

        # Customer tier weights
        tier_weights = {'enterprise': 40, 'premium': 30, 'basic': 20, 'free': 10}
        priority_score += tier_weights.get(self.customer_tier, 10)

        # Business impact weights
        impact_weights = {'critical': 30, 'high': 20, 'medium': 10, 'low': 5}
        priority_score += impact_weights.get(self.business_impact.lower(), 5)

        # Affected users factor
        if self.affected_users > 100:
            priority_score += 20
        elif self.affected_users > 10:
            priority_score += 10

        # Response time urgency
        if self.response_time_hours <= 2:
            priority_score += 10

        # Store computed priority (you'd add this field to the model)
        # self.computed_priority = 'urgent' if priority_score >= 80 else ...

        return self

class SecureTicketData(BaseModel):
    """Model with PII validation and sanitization"""
    customer_name: str
    email: str
    phone: Optional[str] = None
    credit_card_last4: Optional[str] = None
    ssn_last4: Optional[str] = None

    @field_validator('credit_card_last4')
    @classmethod
    def validate_cc_last4(cls, v: Optional[str]) -> Optional[str]:
        if v is None:
            return v
        if not re.match(r'^\d{4}$', v):
            raise ValueError('Credit card last 4 must be exactly 4 digits')
        return v

    @field_validator('phone')
    @classmethod
    def sanitize_phone(cls, v: Optional[str]) -> Optional[str]:
        """Remove all non-digit characters from phone"""
        if v is None:
            return v
        digits_only = re.sub(r'\D', '', v)
        if len(digits_only) < 10:
            raise ValueError('Phone number must have at least 10 digits')
        return digits_only

    @model_validator(mode='after')
    def validate_pii_consistency(self) -> 'SecureTicketData':
        """Ensure PII fields are consistent"""
        # If SSN is provided, credit card should also be provided for identity verification
        if self.ssn_last4 and not self.credit_card_last4:
            raise ValueError('Credit card verification required when SSN is provided')
        return self

用途：为复杂业务需求实现复杂的验证逻辑。

模式:

python

from pydantic import BaseModel, field_validator, model_validator
from typing import Any, Optional
import re

class TicketPrioritization(BaseModel):
    customer_tier: str
    issue_category: str
    response_time_hours: int
    business_impact: str
    affected_users: int = Field(ge=1)

    @field_validator('customer_tier')
    @classmethod
    def validate_tier(cls, v: str) -> str:
        valid_tiers = {'free', 'basic', 'premium', 'enterprise'}
        v = v.lower()
        if v not in valid_tiers:
            raise ValueError(f'Invalid tier. Must be one of: {valid_tiers}')
        return v

    @model_validator(mode='after')
    def calculate_priority(self) -> 'TicketPrioritization':
        """Auto-calculate priority based on multiple factors"""
        priority_score = 0

        # 客户层级权重
        tier_weights = {'enterprise': 40, 'premium': 30, 'basic': 20, 'free': 10}
        priority_score += tier_weights.get(self.customer_tier, 10)

        # 业务影响权重
        impact_weights = {'critical': 30, 'high': 20, 'medium': 10, 'low': 5}
        priority_score += impact_weights.get(self.business_impact.lower(), 5)

        # 受影响用户数因素
        if self.affected_users > 100:
            priority_score += 20
        elif self.affected_users > 10:
            priority_score += 10

        # 响应时间紧急度
        if self.response_time_hours <= 2:
            priority_score += 10

        # 存储计算出的优先级（需在模型中添加此字段）
        # self.computed_priority = 'urgent' if priority_score >= 80 else ...

        return self

class SecureTicketData(BaseModel):
    """Model with PII validation and sanitization"""
    customer_name: str
    email: str
    phone: Optional[str] = None
    credit_card_last4: Optional[str] = None
    ssn_last4: Optional[str] = None

    @field_validator('credit_card_last4')
    @classmethod
    def validate_cc_last4(cls, v: Optional[str]) -> Optional[str]:
        if v is None:
            return v
        if not re.match(r'^\d{4}$', v):
            raise ValueError('Credit card last 4 must be exactly 4 digits')
        return v

    @field_validator('phone')
    @classmethod
    def sanitize_phone(cls, v: Optional[str]) -> Optional[str]:
        """Remove all non-digit characters from phone"""
        if v is None:
            return v
        digits_only = re.sub(r'\D', '', v)
        if len(digits_only) < 10:
            raise ValueError('Phone number must have at least 10 digits')
        return digits_only

    @model_validator(mode='after')
    def validate_pii_consistency(self) -> 'SecureTicketData':
        """Ensure PII fields are consistent"""
        # 若提供了SSN，也需提供信用卡信息用于身份验证
        if self.ssn_last4 and not self.credit_card_last4:
            raise ValueError('Credit card verification required when SSN is provided')
        return self

10. Performance Optimization with Pydantic V2

10. Pydantic V2性能优化

Purpose: Optimize validation performance for high-throughput support systems.

Key Strategies:

Use TypeAdapter for bulk validation:

python

from pydantic import TypeAdapter
from typing import List

用途：为高吞吐量支持系统优化验证性能。

核心策略:

使用TypeAdapter进行批量验证:

python

from pydantic import TypeAdapter
from typing import List

Define adapter once, reuse for validation

定义一次适配器，重复用于验证

ticket_list_adapter = TypeAdapter(List[SupportTicket])

Fast bulk validation

快速批量验证

tickets_data = [...] # List of dictionaries validated_tickets = ticket_list_adapter.validate_python(tickets_data)


2. **Leverage strict mode for performance**:
```python
from pydantic import BaseModel, ConfigDict

class FastTicket(BaseModel):
    model_config = ConfigDict(strict=True)  # No type coercion

    ticket_id: int  # Must be int, won't coerce from string
    priority: str

Use discriminated unions for polymorphic data:

python

from typing import Literal, Union
from pydantic import Field, BaseModel

class EmailTicket(BaseModel):
    ticket_type: Literal['email'] = 'email'
    customer_email: str
    subject: str

class PhoneTicket(BaseModel):
    ticket_type: Literal['phone'] = 'phone'
    phone_number: str
    call_duration: int

class ChatTicket(BaseModel):
    ticket_type: Literal['chat'] = 'chat'
    chat_session_id: str

tickets_data = [...] # 字典列表 validated_tickets = ticket_list_adapter.validate_python(tickets_data)


2. **利用严格模式提升性能**:
```python
from pydantic import BaseModel, ConfigDict

class FastTicket(BaseModel):
    model_config = ConfigDict(strict=True)  # 无类型转换

    ticket_id: int  # 必须是int类型，不会从字符串转换
    priority: str

使用鉴别联合处理多态数据:

python

from typing import Literal, Union
from pydantic import Field, BaseModel

class EmailTicket(BaseModel):
    ticket_type: Literal['email'] = 'email'
    customer_email: str
    subject: str

class PhoneTicket(BaseModel):
    ticket_type: Literal['phone'] = 'phone'
    phone_number: str
    call_duration: int

class ChatTicket(BaseModel):
    ticket_type: Literal['chat'] = 'chat'
    chat_session_id: str

Fast dispatch based on discriminator field

基于鉴别字段快速分发

TicketUnion = Union[EmailTicket, PhoneTicket, ChatTicket]

class TicketProcessor(BaseModel): ticket: TicketUnion = Field(discriminator='ticket_type')


4. **Reuse models and avoid dynamic creation**:
```python

TicketUnion = Union[EmailTicket, PhoneTicket, ChatTicket]

class TicketProcessor(BaseModel): ticket: TicketUnion = Field(discriminator='ticket_type')


4. **复用模型，避免动态创建**:
```python

Good: Define once, reuse

推荐：定义一次，重复使用

class TicketModel(BaseModel): ticket_id: int subject: str

Avoid: Dynamic model creation in loops

避免：在循环中动态创建模型

for data in ticket_data: # Don't create models dynamically pass

undefined

for data in ticket_data: # 不要动态创建模型 pass

undefined

Error Handling Best Practices

错误处理最佳实践

Comprehensive Validation Error Handling

全面的验证错误处理

python

from pydantic import ValidationError
import logging

logger = logging.getLogger(__name__)

def process_ticket_submission(data: dict) -> Optional[SupportTicket]:
    """Process ticket with comprehensive error handling"""
    try:
        ticket = SupportTicket(**data)
        logger.info(f"Ticket {ticket.ticket_id} validated successfully")
        return ticket

    except ValidationError as e:
        # Log detailed validation errors
        logger.error(f"Validation failed for ticket submission: {e.error_count()} errors")

        for error in e.errors():
            field = '.'.join(str(loc) for loc in error['loc'])
            error_type = error['type']
            message = error['msg']

            logger.error(f"Field '{field}': {message} (type: {error_type})")

        # Return user-friendly error response
        return None

    except Exception as e:
        logger.exception(f"Unexpected error processing ticket: {str(e)}")
        return None

python

from pydantic import ValidationError
import logging

logger = logging.getLogger(__name__)

def process_ticket_submission(data: dict) -> Optional[SupportTicket]:
    """Process ticket with comprehensive error handling"""
    try:
        ticket = SupportTicket(**data)
        logger.info(f"Ticket {ticket.ticket_id} validated successfully")
        return ticket

    except ValidationError as e:
        # 记录详细的验证错误
        logger.error(f"Validation failed for ticket submission: {e.error_count()} errors")

        for error in e.errors():
            field = '.'.join(str(loc) for loc in error['loc'])
            error_type = error['type']
            message = error['msg']

            logger.error(f"Field '{field}': {message} (type: {error_type})")

        # 返回用户友好的错误响应
        return None

    except Exception as e:
        logger.exception(f"Unexpected error processing ticket: {str(e)}")
        return None

Testing Pydantic Models

测试Pydantic模型

Unit Test Pattern

单元测试模式

python

import pytest
from pydantic import ValidationError

def test_support_ticket_validation():
    """Test ticket validation logic"""
    # Valid ticket
    valid_data = {
        'ticket_id': 123,
        'customer_email': 'test@example.com',
        'subject': 'Test Issue',
        'description': 'This is a test ticket with enough description',
        'priority': 'medium',
        'created_at': '2024-01-15T10:00:00'
    }
    ticket = SupportTicket(**valid_data)
    assert ticket.ticket_id == 123
    assert ticket.priority == 'medium'

    # Invalid priority
    with pytest.raises(ValidationError) as exc_info:
        SupportTicket(**{**valid_data, 'priority': 'invalid'})

    errors = exc_info.value.errors()
    assert any(e['loc'] == ('priority',) for e in errors)

    # Missing required field
    with pytest.raises(ValidationError):
        incomplete_data = {k: v for k, v in valid_data.items() if k != 'subject'}
        SupportTicket(**incomplete_data)

python

import pytest
from pydantic import ValidationError

def test_support_ticket_validation():
    """Test ticket validation logic"""
    # 有效工单
    valid_data = {
        'ticket_id': 123,
        'customer_email': 'test@example.com',
        'subject': 'Test Issue',
        'description': 'This is a test ticket with enough description',
        'priority': 'medium',
        'created_at': '2024-01-15T10:00:00'
    }
    ticket = SupportTicket(**valid_data)
    assert ticket.ticket_id == 123
    assert ticket.priority == 'medium'

    # 无效优先级
    with pytest.raises(ValidationError) as exc_info:
        SupportTicket(**{**valid_data, 'priority': 'invalid'})

    errors = exc_info.value.errors()
    assert any(e['loc'] == ('priority',) for e in errors)

    # 缺少必填字段
    with pytest.raises(ValidationError):
        incomplete_data = {k: v for k, v in valid_data.items() if k != 'subject'}
        SupportTicket(**incomplete_data)

Guidelines for Customer Support Context

客户支持场景指南

Always validate user input: Never trust data from support forms, API calls, or external systems
Provide helpful error messages: Users should understand what's wrong and how to fix it
Log validation failures: Track patterns in validation errors to improve forms and documentation
Use appropriate field constraints: Balance security with usability
Implement business rule validation: Beyond types, validate business logic
Handle PII carefully: Mask or encrypt sensitive data, exclude from logs
Version your models: Use model versioning for API compatibility
Test edge cases: Include tests for boundary conditions and unusual inputs
Document models thoroughly: Use Field descriptions and JSON schema examples
Monitor validation performance: Track validation times for high-volume operations

始终验证用户输入：永远不要信任来自支持表单、API调用或外部系统的数据
提供有帮助的错误信息：用户应能理解问题所在及解决方法
记录验证失败：跟踪验证错误的模式，以改进表单和文档
使用合适的字段约束：平衡安全性与易用性
实现业务规则验证：除类型检查外，验证业务逻辑
谨慎处理PII数据：屏蔽或加密敏感数据，避免记录到日志中
为模型版本化：使用模型版本保证API兼容性
测试边缘情况：包含边界条件和异常输入的测试
详细记录模型：使用Field描述和JSON Schema示例
监控验证性能：跟踪高流量操作的验证时间

Common Patterns and Anti-Patterns

常见模式与反模式

Pattern: Request/Response Separation

模式：请求/响应分离

python

undefined

python

undefined

Good: Separate models for requests and responses

推荐：为请求和响应使用独立模型

class TicketCreateRequest(BaseModel): subject: str description: str

class TicketResponse(BaseModel): ticket_id: int subject: str description: str created_at: datetime

class TicketCreateRequest(BaseModel): subject: str description: str

class TicketResponse(BaseModel): ticket_id: int subject: str description: str created_at: datetime

Avoid: Using same model for input and output

避免：输入输出使用同一模型

undefined

undefined

Pattern: Default Factory for Mutable Defaults

模式：为可变默认值使用default_factory

python

undefined

python

undefined

Good: Use default_factory

推荐：使用default_factory

class Ticket(BaseModel): tags: list[str] = Field(default_factory=list)

Avoid: Mutable default

避免：可变默认值

class BadTicket(BaseModel): tags: list[str] = [] # Shared across instances!

undefined

class BadTicket(BaseModel): tags: list[str] = [] # 所有实例共享同一个列表！

undefined

Pattern: Computed Fields

模式：计算字段

python

undefined

python

undefined

Good: Use computed_field for derived values

推荐：使用computed_field处理派生值

from pydantic import computed_field

class Ticket(BaseModel): created_at: datetime

@computed_field
@property
def age_days(self) -> int:
    return (datetime.now() - self.created_at).days

undefined

from pydantic import computed_field

class Ticket(BaseModel): created_at: datetime

@computed_field
@property
def age_days(self) -> int:
    return (datetime.now() - self.created_at).days

undefined

Skill Application Checklist

技能应用检查清单

References and Resources

参考资料

Pydantic V2 Documentation: https://docs.pydantic.dev/latest/
Pydantic Settings: https://docs.pydantic.dev/latest/concepts/pydantic_settings/
FastAPI Integration: https://fastapi.tiangolo.com/tutorial/body/
Migration Guide (V1 to V2): https://docs.pydantic.dev/latest/migration/
Performance Tips: https://docs.pydantic.dev/latest/concepts/performance/

Pydantic V2文档：https://docs.pydantic.dev/latest/
Pydantic Settings：https://docs.pydantic.dev/latest/concepts/pydantic_settings/
FastAPI集成：https://fastapi.tiangolo.com/tutorial/body/
迁移指南（V1到V2）：https://docs.pydantic.dev/latest/migration/
性能技巧：https://docs.pydantic.dev/latest/concepts/performance/