# Pydantic 数据验证技能 ## 概述您是一名专门从事客户支持系统数据验证的 Pydantic 专家。您的角色是帮助构建强大的、类型安全的数据模型，以使用 Pydantic V2 验证支持票证、用户数据、API 请求和配置设置。 ## 核心能力 ### 1\.基础模型基础知识 **目的**：通过自动类型强制和全面的错误报告创建经过验证的数据模型。 **关键原则**： * 使用Python类型提示定义模型 * 利用实例化的自动验证 * 使用`model_dump()`和`model_dump_json()`进行序列化 * 优雅地处理`ValidationError`异常 * 为支持操作实施正确的错误日志记录 **基本模式**：从 pydantic 导入 BaseModel、Field、ValidationError 从日期时间导入日期时间从输入 import 可选类支持票（基础模型）：票证 ID：整数客户电子邮件：str 主题：str = 字段(min_length=5, max_length=200) 描述：str = 字段(min_length=20) 优先级：str = Field(pattern=r'^(低|中|高|紧急)$') 创建时间：日期时间分配给：可选[str] =无状态：str = '打开' 尝试：票证 = 支持票证( Ticket_id=12345, customer_email='customer@example.com', subject='登录问题', description='密码重置后无法访问我的帐户', 优先级='高', 创建于='2024-01-15T10:30:00' ）打印(ticket.model_dump()) 除了 ValidationError 为 e： # 记录验证错误以供支持团队审核对于 e.errors() 中的错误： print(f"字段：{error['loc']}，错误：{error['msg']}") ### 2\.现场配置和限制 **目的**：将精细的验证规则应用于各个字段以保证数据质量。 **共同约束**： * **字符串验证** : `min_length`, `max_length`, `pattern`, `strip_whitespace` * **数字验证** : `gt`, `ge`, `lt`, `le`, `multiple_of` * **字段元数据** : `title`, `description`, `examples`, `json_schema_extra` * **序列化控制** ：`alias`、`serialization_alias`、`exclude`、`include` **客户支持示例**：从 pydantic 导入 BaseModel、Field、EmailStr、HttpUrl 从输入导入注释从日期时间导入日期时间类 CustomerProfile(BaseModel)： # 有约束的 ID 字段 customer_id：注释[int，Field（gt = 0，description =“唯一客户标识符”）] # 联系信息及验证电子邮件：EmailStr 电话: Annotated[str, Field(pattern=r'^\+?1?\d{9,15}$', description="国际电话格式")] # 具有长度限制的名称字段 first_name: 带注释的[str, Field(min_length=1, max_length=50, strip_whitespace=True)] last_name: 带注释的[str, Field(min_length=1, max_length=50, strip_whitespace=True)] # 公司信息（可选）公司名称：可选[注释[str，字段(max_length = 100)]] =无公司网站：可选[HttpUrl] = 无 # 默认支持层 support_tier: 带注释的[str, Field(pattern=r'^(basic|premium|enterprise)$')] = 'basic' # 元数据字段注册日期：日期时间 last_contact: 可选[日期时间] = 无注释：str = Field(default='', max_length=2000, description="内部注释") # 从序列化中排除（仅供内部使用）内部得分：int = 字段（默认=0，排除=True）模型配置 = { 'str_strip_whitespace'：正确， “验证分配”：正确， “按名称填充”：正确 } ### 3\.自定义字段验证器 **目的**：实现基本类型检查之外的业务逻辑验证。 **@field_validator 模式**：从 pydantic 导入 BaseModel、field_validator、ValidationInfo 进口再类 SupportTicketSubmission(BaseModel)：客户电子邮件：str 主题：str 描述：str 类别： STR 附件：列表[str] = [] @field_validator('customer_email') @类方法 def validate_email_domain(cls, v: str) -> str: """验证电子邮件格式并检查被阻止的域""" blocked_domains = ['tempmail.com', 'throtaway.email'] 如果 '@' 不在 v 中：引发 ValueError('电子邮件格式无效') 域 = v.split('@')[1].lower() 如果域位于blocked_domains中： raise ValueError(f'电子邮件域 {domain} 不允许') 返回 v.lower() @field_validator('主题') @类方法 def validate_subject(cls, v: str) -> str: """确保主题有意义且不是垃圾邮件""" v = v.strip() # 检查最小字数单词 = v.split() 如果 len(字数) < 2: raise ValueError('主题必须包含至少 2 个单词') # 检查垃圾邮件模式 spam_patterns = [r'伟哥', r'赌场', r'彩票'] 对于 spam_patterns 中的模式： if re.search(pattern, v, re.IGNORECASE): raise ValueError('主题包含禁止内容') 返回v @field_validator('附件') @类方法 def validate_attachments(cls, v: list[str]) -> list[str]: """验证附件文件扩展名""" allowed_extensions = {'.pdf', '.jpg', '.jpeg', '.png', '.doc', '.docx', '.txt'} 对于 v 中的文件名： ext = 文件名[文件名.rfind('.'):].lower() if '.'在文件名 else '' 中如果 ext 不在 allowed_extensions 中： raise ValueError(f'不允许文件类型 {ext}。允许：{allowed_extensions}') 如果 len(v) > 5： raise ValueError('最多允许 5 个附件') 返回v @field_validator('类别') @类方法 def validate_category(cls, v: str) -> str: """规范并验证票证类别""" 有效类别 = { '技术'、'计费'、'帐户'、'feature_request'、 '错误报告'，'一般查询' } v_normalized = v.lower().replace(' ', '_') 如果 v_normalized 不在 valid_categories 中： raise ValueError(f'无效类别。有效选项：{valid_categories}') 返回 v_normalized ### 4\.模型级验证 **目的**：验证多个字段之间的关系并执行跨字段验证。 **@model_validator 模式**：从 pydantic 导入 BaseModel、model_validator、ValidationError 从日期时间导入日期时间，时间增量输入 import Any，可选类 TicketSchedule(BaseModel): 票证 ID：整数预定开始时间：日期时间预定结束时间：日期时间 Technical_id：可选[int] = None 预计小时数：浮动优先级：str @model_validator(模式='之前') @类方法 def preprocess_data(cls, data: Any) -> Any: """在现场验证之前对数据进行预处理和标准化""" 如果是实例（数据，字典）： # 如果没有提供，自动生成预计时间如果“scheduled_start”在数据中，“scheduled_end”在数据中，“estimated_hours”不在数据中： start = datetime.fromisoformat(data['scheduled_start']) end = datetime.fromisoformat(data['scheduled_end']) 数据['估计小时'] = (结束 - 开始).total_seconds() / 3600 # 标准化优先级如果数据中的“优先级”：数据['优先级'] = 数据['优先级'].lower() 返回数据 @model_validator(模式='之后') def validate_schedule(self) -> 'TicketSchedule': """验证所有字段后验证调度逻辑""" # 确保结束在开始之后如果 self.scheduled_end <= self.scheduled_start： raise ValueError('scheduled_end 必须在 Scheduled_start 之后') # 根据优先级验证持续时间持续时间 = (self.scheduled_end - self.scheduled_start).total_seconds() / 3600 如果 self.priority == '紧急' 且持续时间 > 2： raise ValueError('紧急票必须在2小时内安排') 如果 self.priority == 'low' 且持续时间 < 1： raise ValueError('低优先级门票需要至少 1 小时分配') # 确保预计时间与持续时间匹配如果 abs(self.estimated_hours - 持续时间) > 0.1: raise ValueError('estimated_hours 必须与预定的持续时间匹配') # 验证营业时间（上午 9 点至下午 6 点）如果 self.scheduled_start.hour < 9 或 self.scheduled_start.hour >= 18： raise ValueError('门票必须在营业时间（上午 9 点 - 下午 6 点）安排') 返回自我票务升级类（BaseModel）：票证 ID：整数当前受让人：str escalation_level：整数原因：str 请求者：str 批准者：可选[str] =无 @model_validator(模式='之后') def validate_escalation_approval(self) -> 'TicketEscalation': """确保高层升级需要批准""" 如果 self.escalation_level >= 3 并且不是 self.approved_by： raise ValueError('3级以上升级需要经理批准') 如果 self.approved_by == self.requested_by： raise ValueError('批准者必须与请求者不同') 返回自我 ### 5\.嵌套模型和复杂结构 **目的**：为复杂的支持工作流程构建分层数据模型。 **图案**：从 pydantic 导入 BaseModel、Field 从日期时间导入日期时间 from 输入 import 可选，文字从枚举导入枚举类 TicketStatus(str, Enum): OPEN = '打开' IN_PROGRESS = '进行中' PENDING_CUSTOMER = '待定客户' 已解决 = '已解决' 关闭 = '关闭' 类注释（基础模型）：评论 ID：int 作者：str 内容：str = 字段(min_length=1, max_length=5000) 时间戳：日期时间 is_internal: 布尔 = False 类附件（基础模型）：文件名：str file_size_bytes: int = Field(gt=0, le=10_000_000) # 最大 10MB 内容类型：str 上传者：str uploaded_at: 日期时间网址：str 类解析（BaseModel）：解析者：str 决议日期：日期时间 resolution_summary: str = 字段(min_length=20, max_length=2000) root_cause: 可选[str] = 无预防措施：可选[str] =无 customer_satisfaction_score：可选[int] =字段（默认=无，ge=1，le=5）类 CustomerInfo(BaseModel)：客户 ID：整数名称：str 电子邮件：str 电话：可选[str] = 无 account_type: Literal['免费', '基本', '高级', '企业'] 注册日期：日期时间类 CompleteTicket(BaseModel): # 核心门票信息票证 ID：整数客户：客户信息 # 门票详情主题：str = 字段(min_length=5, max_length=200) 描述：str = 字段(min_length=20) 类别： STR 优先级：str 状态：票证状态 # 分配和跟踪分配给：可选[str] =无创建时间：日期时间更新时间：日期时间 # 内容丰富评论：列表[评论] = [] 附件：列表[附件] = [] # 解决方案（如果已解决/关闭）分辨率：可选[分辨率] = 无 # 元数据标签： list[str] = Field(default_factory=list, max_length=10) related_tickets: 列表[int] = 字段(default_factory=列表) @model_validator(模式='之后') def validate_ticket_state(self) -> 'CompleteTicket': """确保票据状态一致""" 如果 self.status 在 (TicketStatus.RESOLVED, TicketStatus.CLOSED) 中而不是 self.resolution： raise ValueError('已解决/已关闭的票证必须包含解决方案详细信息') 如果 self.status == TicketStatus.IN_PROGRESS 而不是 self.assigned_to： raise ValueError('必须分配正在进行的票证') 返回自我模型配置 = { 'use_enum_values'：正确， “验证分配”：正确， 'json_schema_extra'：{ '例子'：[{ 'ticket_id': 12345, “客户”：{ “客户 ID”：6789， '姓名': '约翰·多伊', '电子邮件': 'john@example.com', 'account_type': '高级', '注册日期': '2023-01-01T00:00:00' }, 'subject': '无法访问仪表板', 'description': '尝试访问分析仪表板时出现 403 错误', '类别': '技术', “优先级”：“高”， '状态': '进行中', '分配给'：'tech_support_1'， '创建时间': '2024-01-15T10:00:00', 'updated_at': '2024-01-15T11:30:00' }] } } ### 6\.请求/响应模型的 FastAPI 集成 **目的**：通过自动验证和文档创建类型安全的 API 端点。 **图案**：从 fastapi 导入 FastAPI、HTTPException、状态从 pydantic 导入 BaseModel、Field、ValidationError 输入 import 可选、列表从日期时间导入日期时间 app = FastAPI(title="支持票证 API") # 请求模型类 TicketCreateRequest(BaseModel): 客户电子邮件：str 主题：str = 字段(min_length=5, max_length=200) 描述：str = 字段(min_length=20, max_length=5000) 类别： STR 优先级：str = Field(默认='中',pattern=r'^(低|中|高|紧急)$') 附件：List[str] = Field(default_factory=list, max_length=5) 模型配置 = { 'json_schema_extra'：{ '例子'：[{ 'customer_email': 'user@example.com', 'subject': '无法重置密码', 'description': '我点击了重置密码链接，但没有收到电子邮件', '类别': '账户', “优先级”：“高”， “附件”：[] }] } } 类 TicketUpdateRequest(BaseModel): 主题：可选[str] =字段（默认=无，min_length=5，max_length=200）描述：可选[str] =字段（默认=无，min_length=20）状态：可选[str] =无分配给：可选[str] =无优先级：可选[str] =字段（默认=无，模式=r'^（低|中|高|紧急）$'） # 响应模型类 TicketResponse(BaseModel): 票证 ID：整数客户电子邮件：str 主题：str 描述：str 类别： STR 优先级：str 状态：str 分配给：可选[str] 创建时间：日期时间更新时间：日期时间 model_config = {'from_attributes': True} # 用于 ORM 兼容性类 TicketListResponse(BaseModel): 票证：列表[TicketResponse] 总计：整数页面：int 页面大小：int 类错误响应（基础模型）：错误代码：str 消息：str 详细信息：可选[dict] =无 # API 端点 @应用程序.post（ “/门票/”， response_model=TicketResponse, status_code=status.HTTP_201_CREATED, 回复={ 400: {"model": ErrorResponse, "description": "验证错误"}, 500：{“模型”：ErrorResponse，“描述”：“服务器错误”} } ）异步 def create_ticket(ticket: TicketCreateRequest): """创建具有自动验证功能的新支持票证""" 尝试： # 这里的业务逻辑 # Ticket_data 由 Pydantic 自动验证新票 = { “ticket_id”：12345， "customer_email": 票证.customer_email, “主题”：票证.主题， “描述”：票证.描述， “类别”：票证.类别， “优先级”：ticket.priority， “状态”：“开放”， “分配给”：无， “创建时间”：datetime.now(), “updated_at”：datetime.now() } 返回 TicketResponse(**new_ticket) 除了异常 e：引发 HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR，详细信息 = {“error_code”：“CREATION_FAILED”，“消息”：str（e）} ） @app.get("/tickets/", response_model=TicketListResponse) 异步 def list_tickets( 页面：int = 字段（默认=1，ge=1）， page_size: int = Field(默认=20, ge=1, le=100), 状态：可选[str] =无，优先级：可选[str] =无）： """列出带有分页和过滤的票证""" # 查询逻辑在这里 Tickets = [] # 从数据库中获取返回工单列表响应（门票=门票，总计=0，页=页，页大小=页大小） @app.patch("/tickets/{ticket_id}", response_model=TicketResponse) 异步 def update_ticket(ticket_id: int, update: TicketUpdateRequest): """通过部分验证更新票证""" # 仅验证提供的字段 update_data = update.model_dump(exclude_unset=True) # 将更新应用到数据库 # 返回更新后的票证通过 ### 7\.配置管理的基本设置 **用途**：通过验证从环境变量管理应用程序配置。 **图案**：从 pydantic_settings 导入 BaseSettings、SettingsConfigDict 从 pydantic 导入 Field、PostgresDsn、RedisDsn、EmailStr 输入 import 可选、列表类数据库设置（BaseSettings）： """数据库配置""" 主机：str = Field（默认='localhost'，描述='数据库主机'）端口：int = 字段（默认=5432，ge=1，le=65535）用户名：str 密码：str 数据库：str = Field（默认='support_db'） # 连接字符串的计算属性 @属性 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' ）类 RedisSettings(BaseSettings): """Redis缓存配置""" url: RedisDsn = Field(默认='redis://localhost:6379/0') 最大连接数：int = 字段（默认=50，ge=1） socket_timeout: int = Field(默认=5, ge=1) model_config = SettingsConfigDict( env_prefix='REDIS_', env_file='.env' ）类电子邮件设置（基本设置）： """电子邮件通知配置""" smtp_host: str = 字段(默认='smtp.gmail.com') smtp_port：int = 字段（默认=587，ge=1，le=65535） smtp_用户名：str smtp_密码：str from_email: 电子邮件地址 support_email: 电子邮件Str use_tls: bool = Field(默认=True) model_config = SettingsConfigDict( env_prefix='EMAIL_', env_file='.env' ）类支持设置（基本设置）： """支持系统配置""" max_tickets_per_page: int = Field(默认=50, ge=1, le=200) max_attachment_size_mb：int = 字段（默认=10，ge=1，le=100） allowed_file_extensions: 列表[str] = 字段( 默认=['.pdf', '.jpg', '.jpeg', '.png', '.doc', '.docx'] ） auto_assign_enabled: bool = Field(默认=True) escalation_hours：int = Field（默认=24，ge=1） Business_hours_start：int = 字段（默认=9，ge=0，le=23） Business_hours_end：int = 字段（默认=18，ge=0，le=23） model_config = SettingsConfigDict( env_prefix='SUPPORT_', env_file='.env' ）类应用程序设置（基础设置）： """主要应用程序设置""" app_name: str = Field(默认='支持票务系统') app_version: str = Field(默认='1.0.0') 调试：bool = Field（默认=False） api_key: 字符串秘密密钥：str allowed_origins: List[str] = Field(default=['http://localhost:3000']) # 嵌套设置数据库：DatabaseSettings = Field(default_factory=DatabaseSettings) redis: RedisSettings = 字段(default_factory=RedisSettings) 电子邮件：EmailSettings = 字段（default_factory=EmailSettings）支持：SupportSettings = Field(default_factory=SupportSettings) model_config = SettingsConfigDict( env_file='.env', env_file_encoding='utf-8', 区分大小写=假，额外='忽略' ） # 用法设置=应用程序设置（） print(f"正在连接到数据库：{settings.database.connection_url}") print(f"每页最大票数：{settings.support.max_tickets_per_page}") ### 8\.序列化和反序列化 **用途**：控制模型与字典、JSON 和其他格式之间的转换方式。 **图案**：从 pydantic 导入 BaseModel、Field、field_serializer、compute_field 从日期时间导入日期时间从输入 import 可选类 TicketExport(BaseModel): 票证 ID：整数客户电子邮件：str 主题：str 创建时间：日期时间状态：str 优先级：str 分配给：可选[str] =无内部注释：str = 字段（默认=''，排除=True） @field_serializer('customer_email') def mask_email(self, email: str) -> str: """在导出时屏蔽电子邮件以保护隐私""" 如果电子邮件中包含“@”：本地，域= email.split（'@'） masked = local[:2] + '***' + local[-1:] if len(local) > 3 else '***' 返回 f"{masked}@{domain}" 返回电子邮件 @field_serializer('created_at') def format_datetime(self, dt: 日期时间) -> str: """导出日期时间格式""" return dt.strftime('%Y-%m-%d %H:%M:%S') @计算字段 @属性 def days_open(self) -> int： """计算票证创建后的天数""" 返回 (datetime.now() - self.created_at).days # 序列化模式票证 = 票证导出( 票证 ID=123, customer_email='john.doe@example.com', subject='登录问题', 创建时间=日期时间.now(), 状态='打开', 优先级='高', Internal_notes='客户拨打了两次' ） # 标准序列化打印(ticket.model_dump()) # {'ticket_id': 123, 'customer_email': 'jo***e@example.com', ...} # 包含所有字段（甚至排除）打印（ticket.model_dump（模式='python'，exclude_none=False）） # 序列化为 JSON json_str = Ticket.model_dump_json(缩进=2) 打印（json_str） # 排除特定字段打印（ticket.model_dump（排除= {'internal_notes'，'assigned_to'}）） # 仅包含特定字段 print(ticket.model_dump(include={'ticket_id', 'subject', 'status'})) ### 9\.先进的验证技术 **目的**：针对复杂的业务需求实现复杂的验证逻辑。 **图案**：从 pydantic 导入 BaseModel、field_validator、model_validator 输入 import Any，可选进口再类 TicketPrioritization(BaseModel)：客户层：str 问题类别：str 响应时间小时：int 业务影响：str 受影响的用户：int = 字段(ge=1) @field_validator('customer_tier') @类方法 def validate_tier(cls, v: str) -> str: valid_tiers = {'免费', '基本', '高级', '企业'} v = v.lower() 如果 v 不在 valid_tiers 中： raise ValueError(f'无效层。必须是以下之一：{valid_tiers}') 返回v @model_validator(模式='之后') defcalculate_priority(self) -> 'TicketPrioritization': """根据多种因素自动计算优先级""" 优先级分数 = 0 # 客户等级权重 tier_weights = {'企业'：40，'高级'：30，'基本'：20，'免费'：10} priority_score += tier_weights.get(self.customer_tier, 10) # 业务影响权重 Impact_weights = {'关键'：30，'高'：20，'中'：10，'低'：5} priority_score += Impact_weights.get(self.business_impact.lower(), 5) # 受影响的用户因素如果 self.affected_users > 100：优先级分数 += 20 elif self.affected_users > 10：优先级分数 += 10 # 响应时间紧迫性如果 self.response_time_hours <= 2: 优先级分数 += 10 # 存储计算出的优先级（您可以将此字段添加到模型中） # self.compated_priority = '紧急' 如果priority_score >= 80 else ... 返回自我类 SecureTicketData(BaseModel)： """具有 PII 验证和清理的模型""" 客户名称：str 电子邮件：str 电话：可选[str] = 无 Credit_card_last4：可选[str] =无 ssn_last4：可选[str] =无 @field_validator('credit_card_last4') @类方法 def validate_cc_last4(cls, v: 可选[str]) -> 可选[str]: 如果 v 为无：返回v 如果不是 re.match(r'^\d{4}$', v): raise ValueError('信用卡最后4位必须正好是4位') 返回v @field_validator('电话') @类方法 def sanitize_phone(cls, v: 可选[str]) -> 可选[str]: """从手机中删除所有非数字字符""" 如果 v 为无：返回v digitals_only = re.sub(r'\D', '', v) 如果 len(digits_only) < 10: raise ValueError('电话号码必须至少有 10 位数字') 仅返回数字 @model_validator(模式='之后') def validate_pii_consistency(self) -> 'SecureTicketData': """确保 PII 字段一致""" # 如果提供SSN，还应提供信用卡以进行身份验证如果 self.ssn_last4 而不是 self.credit_card_last4： raise ValueError('提供 SSN 时需要信用卡验证') 返回自我 ### 10\.使用 Pydantic V2 进行性能优化 **目的**：优化高通量支持系统的验证性能。 **关键策略**： 1. **使用TypeAdapter进行批量验证**：从 pydantic 导入 TypeAdapter 从输入导入列表 # 定义适配器一次，重复使用进行验证 Ticket_list_adapter = TypeAdapter(列表[SupportTicket]) # 快速批量验证 Tickets_data = [...] # 字典列表 valid_tickets = Ticket_list_adapter.validate_python(tickets_data) 2. **利用严格模式提高性能**：从 pydantic 导入 BaseModel、ConfigDict FastTicket 类（基础模型）： model_config = ConfigDict(strict=True) # 无类型强制 Ticket_id: int # 必须是 int，不会从字符串强制转换优先级：str 3. **对多态数据使用可区分联合**：从输入 import Literal, Union 从 pydantic 导入字段，BaseModel 类 EmailTicket(BaseModel): Ticket_type: Literal['email'] = 'email' 客户电子邮件：str 主题：str PhoneTicket 类（基础模型）： Ticket_type: Literal['phone'] = 'phone' 电话号码：str call_duration：整数类 ChatTicket(BaseModel): Ticket_type: Literal['聊天'] = '聊天' chat_session_id：str # 基于鉴别器字段的快速调度 TicketUnion = Union[EmailTicket、PhoneTicket、ChatTicket] 类 TicketProcessor(BaseModel): 票证： TicketUnion = Field(discriminator='ticket_type') 4. **重用模型并避免动态创建** : # 好：定义一次，重复使用类 TicketModel(BaseModel): 票证 ID：整数主题：str # 避免：循环中的动态模型创建对于 Ticket_data 中的数据： # 不要动态创建模型通过 ## 错误处理最佳实践 ### 全面的验证错误处理从 pydantic 导入验证错误导入日志记录记录器=logging.getLogger(__name__) def process_ticket_submission(data: dict) -> 可选[SupportTicket]: """通过全面的错误处理处理票证""" 尝试：票证 = SupportTicket(**数据) logger.info(f"票证 {ticket.ticket_id} 验证成功") 回程票除了 ValidationError 为 e： # 记录详细的验证错误 logger.error(f"票证提交验证失败：{e.error_count()} 错误") 对于 e.errors() 中的错误： field = '.'.join(str(loc) for loc in error['loc']) error_type = 错误['类型'] 消息 = 错误['msg'] logger.error(f"字段'{field}': {message} (类型: {error_type})") # 返回用户友好的错误响应返回无除了异常 e： logger.exception(f"处理票证时出现意外错误：{str(e)}") 返回无 ## 测试 Pydantic 模型 ### 单元测试模式导入pytest 从 pydantic 导入验证错误 def test_support_ticket_validation(): """测试票据验证逻辑""" # 有效门票有效数据 = { “票号”：123， 'customer_email': 'test@example.com', '主题'：'测试问题'， 'description': '这是一张有足够描述的测试票', “优先级”：“中”， '创建时间': '2024-01-15T10:00:00' } 票证 = SupportTicket(**valid_data) 断言 Ticket.ticket_id == 123 断言 Ticket.priority == 'medium' # 优先级无效使用 pytest.raises(ValidationError) 作为 exc_info： SupportTicket(**{**valid_data, '优先级': '无效'}) 错误 = exc_info.value.errors() 断言任何(e['loc'] == ('优先级',) for e inErrors) # 缺少必填字段与 pytest.raises(ValidationError): incomplete_data = {k: v for k, v in valid_data.items() if k != 'subject'} 支持票（**不完整_数据） ## 客户支持环境指南 1. **始终验证用户输入**：永远不要信任来自支持表单、API 调用或外部系统的数据 2. **提供有用的错误消息**：用户应该了解错误所在以及如何修复它 3. **记录验证失败**：跟踪验证错误中的模式以改进表单和文档 4. **使用适当的字段约束**：平衡安全性与可用性 5. **实施业务规则验证**：超越类型，验证业务逻辑 6. **小心处理 PII**：屏蔽或加密敏感数据，从日志中排除 7. **对模型进行版本控制**：使用模型版本控制来实现 API 兼容性 8. **测试边缘情况**：包括边界条件和异常输入的测试 9. **彻底的文档模型**：使用字段描述和 JSON 模式示例 10. **监控验证性能**：跟踪大批量操作的验证时间 ## 常见模式和反模式 ### 模式：请求/响应分离 # 好：请求和响应的独立模型类 TicketCreateRequest(BaseModel): 主题：str 描述：str 类 TicketResponse(BaseModel): 票证 ID：整数主题：str 描述：str 创建时间：日期时间 # 避免：使用相同的模型进行输入和输出 ### 模式：可变默认值的默认工厂 # 好：使用default_factory 票证类（基础模型）：标签： list[str] = Field(default_factory=list) # 避免：可变默认值类 BadTicket(BaseModel): Tags: list[str] = [] # 跨实例共享！ ### 模式：计算字段 # 好：使用计算字段作为派生值从 pydantic 导入计算字段票证类（基础模型）：创建时间：日期时间 @计算字段 @属性 defage_days(self) -> int: 返回 (datetime.now() - self.created_at).days ## 技能申请清单使用此技能时，请确保： * 定义清晰的、带有适当约束的类型化模型 * 为业务逻辑实现自定义验证器 * 优雅地处理ValidationError异常 * 针对不同的上下文（API、数据库、日志）使用适当的序列化 * 适当配置模型（冻结、验证分配等） * 为验证逻辑编写全面的测试 * 包含描述和示例的文档模型 * 使用BaseSettings进行配置管理 * 优化大批量操作的性能 * 遵循 PII 和敏感数据的安全最佳实践 ## 参考资料和资源 * Pydantic V2 文档： * Pydantic 设置： * FastAPI集成： * 迁移指南（V1 到 V2）： * 性能提示：

迂腐的

包含技能