如何在Pydantic模型验证失败时将指定字段设为None?
解决Pydantic模型非关键字段验证失败的兼容问题
问题背景
我定义了如下Pydantic模型:
from datetime import datetime from typing import Optional from pydantic import BaseModel, Field class InvalidTransaction(BaseModel): error_ts: datetime = Field(default_factory=datetime.utcnow) transaction_ts: Optional[datetime] = None filename: str index: int id: Optional[int] = None message: str
其中transaction_ts和id字段从文件读取,格式无法保证,但它们不是关键字段。我需要在这两个字段验证不通过时,仍能正常创建模型对象,并自动将验证失败的字段设为None。
当前测试代码执行后会抛出验证错误:
print(InvalidTransaction(**{ 'transaction_ts': '2022-03 12:30:21', # 无效的datetime格式 'filename': 'myfile.csv', 'index': 3, 'id': 4, 'message': 'Invalid transaction found' }).json())
pydantic.error_wrappers.ValidationError: 1 validation error for InvalidTransaction
transaction_ts
invalid datetime format (type=value_error.datetime)
期望得到的输出:
{"error_ts": "2022-08-11T17:34:36.837724", "transaction_ts": null, "filename": "myfile.csv", "index": 3, "id": 4, "message": "Invalid transaction found"}
解决方案
方法一:给字段添加容错验证器
通过@field_validator给目标字段设置前置验证逻辑,捕获转换失败的异常并返回None:
from datetime import datetime from typing import Optional, Any from pydantic import BaseModel, Field, field_validator class InvalidTransaction(BaseModel): error_ts: datetime = Field(default_factory=datetime.utcnow) transaction_ts: Optional[datetime] = None filename: str index: int id: Optional[int] = None message: str @field_validator('transaction_ts', mode='before') def parse_transaction_ts(cls, v: Any) -> Optional[datetime]: if v is None: return None try: return datetime.fromisoformat(v) except (ValueError, TypeError): return None @field_validator('id', mode='before') def parse_id(cls, v: Any) -> Optional[int]: if v is None: return None try: return int(v) except (ValueError, TypeError): return None
这样即使transaction_ts传入无效格式、id传入非整数内容,都会自动转为None,不会触发模型验证失败。
方法二:自定义可复用的容错类型
如果多个模型或字段需要类似容错逻辑,可以封装自定义类型:
from datetime import datetime from typing import Optional, Any, TypeVar from pydantic import GetCoreSchemaHandler from pydantic_core import core_schema T = TypeVar('T') class TolerantOptional: @classmethod def __get_pydantic_core_schema__(cls, source_type: Any, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema: base_schema = handler(source_type.__args__[0]) return core_schema.no_info_wrap_validator_function( cls.validate, base_schema, serialization=core_schema.plain_serializer_function_ser(lambda x: x), ) @classmethod def validate(cls, value: Any) -> Optional[T]: try: return value except (ValueError, TypeError): return None # 在模型中使用自定义类型 class InvalidTransaction(BaseModel): error_ts: datetime = Field(default_factory=datetime.utcnow) transaction_ts: TolerantOptional[datetime] = None filename: str index: int id: TolerantOptional[int] = None message: str
这种方式把容错逻辑封装起来,能在多个场景重复使用,代码更简洁。
方法三:手动预处理输入数据(简单场景适用)
如果字段数量少,也可以在传入模型前手动处理数据:
input_data = { 'transaction_ts': '2022-03 12:30:21', 'filename': 'myfile.csv', 'index': 3, 'id': 'not_an_int', 'message': 'Invalid transaction found' } # 处理transaction_ts字段 try: input_data['transaction_ts'] = datetime.fromisoformat(input_data['transaction_ts']) except (ValueError, TypeError): input_data['transaction_ts'] = None # 处理id字段 try: input_data['id'] = int(input_data['id']) except (ValueError, TypeError): input_data['id'] = None # 创建模型并输出结果 obj = InvalidTransaction(**input_data) print(obj.json())
这种方式不需要修改模型定义,适合临时快速解决问题。
内容的提问来源于stack exchange,提问作者Luiscri
相关产品推荐
相关产品推荐

