Pydantic校验异常处理与批量数据校验优化问题咨询

背景说明 #

开发校验JSON数据的应用，待校验条目在500-2000条之间，Pydantic完全能胜任这个量级的校验需求，以下针对你的问题给出具体解决方案：

问题解答 #

1. 捕获ValidationError时获取原始对象数据 #

在捕获异常的except块中，直接使用当前循环的food变量即可获取触发异常的原始数据，将其与错误信息一起记录即可用于后续追溯：

valid_foods = []
invalid_foods_with_errors = []

for food in foods:
    try:
        valid_foods.append(Food(**food))
    except pydantic.ValidationError as err:
        # 记录原始数据和错误信息
        invalid_foods_with_errors.append({"data": food, "error": str(err)})
        print(f"无效数据: {food}\n错误信息: {err}", '-'*30, sep='\n')

如果需要更结构化的错误信息，可以使用err.errors()方法获取详细的错误字段和原因，方便后续分析。

2. 分离有效与无效条目 #

Pydantic没有直接提供批量分离的内置方法，但可以通过一次遍历同时收集有效和无效条目，写法简洁高效：

valid_foods = []
invalid_foods = []

for food in foods:
    try:
        valid_foods.append(Food(**food))
    except pydantic.ValidationError:
        invalid_foods.append(food)

# 输出结果
print("有效条目:", valid_foods, sep='\n')
print("-"*30)
print("无效条目:", invalid_foods, sep='\n')

如果需要保留错误信息，参照问题1的写法，将无效数据和错误绑定存储即可。

3. 更符合Python风格的写法（结合try-except的列表推导替代方案） #

列表推导本身无法直接嵌入try-except，但可以封装一个辅助函数，再结合列表推导或生成器表达式实现更简洁的写法：

方案1：辅助函数+列表推导

def validate_food(food_data):
    try:
        return Food(**food_data), None
    except pydantic.ValidationError as err:
        return None, (food_data, err)

# 一次遍历拆分有效和无效结果
valid_foods, invalid_results = zip(*[validate_food(f) for f in foods])

# 过滤掉None值
valid_foods = [item for item in valid_foods if item is not None]
invalid_foods_with_errors = [item for item in invalid_results if item is not None]

方案2：生成器表达式+一次遍历

def process_foods(food_list):
    valid = []
    invalid = []
    for food in food_list:
        try:
            valid.append(Food(**food))
        except pydantic.ValidationError:
            invalid.append(food)
    return valid, invalid

valid_foods, invalid_foods = process_foods(foods)

这种写法既保持了Python的简洁性，又避免了列表推导遇到异常直接终止的问题，同时代码可读性更高。

完整优化代码示例 #

import pydantic

class Food(pydantic.BaseModel):
    name: str
    is_dry: bool
    price: float

    @pydantic.validator('name')
    def validate_name(cls, value: str) -> str:
        if not value.startswith('XA_'):
            raise ValueError('Item is not part of inventory and is skipped.')
        return value

# 模拟加载的JSON数据
foods = [
    {'name': 'XA_APPLE', 'is_dry': True, 'price': 0.30},
    {'name': 'FISH', 'is_dry': False, 'price': 3.30},
    {'name': 'XA_RICE', 'is_dry': True, 'price': 0.50},
    {'name': 'FISH', 'is_dry': False, 'price': 3.30},
]

# 处理数据
valid_foods, invalid_foods_with_errors = [], []
for food in foods:
    try:
        valid_foods.append(Food(**food))
    except pydantic.ValidationError as err:
        invalid_foods_with_errors.append({"data": food, "error": err.errors()})

# 输出结果
print("有效条目:", valid_foods, sep='\n')
print("-"*30)
print("无效条目及错误:")
for item in invalid_foods_with_errors:
    print(f"数据: {item['data']}")
    print(f"错误详情: {item['error']}")
    print("-"*20)

输出结果 #

有效条目:
[Food(name='XA_APPLE', is_dry=True, price=0.3), Food(name='XA_RICE', is_dry=True, price=0.5)]
------------------------------
无效条目及错误:
数据: {'name': 'FISH', 'is_dry': False, 'price': 3.3}
错误详情: [{'loc': ('name',), 'msg': 'Item is not part of inventory and is skipped.', 'type': 'value_error'}]
--------------------
数据: {'name': 'FISH', 'is_dry': False, 'price': 3.3}
错误详情: [{'loc': ('name',), 'msg': 'Item is not part of inventory and is skipped.', 'type': 'value_error'}]
--------------------

内容的提问来源于stack exchange，提问作者Raymond C.