如何跳过无效数据的Pydantic BaseModel创建、记录错误及优化建模

完整实现代码 #

import pandas as pd
from pydantic import BaseModel, Field, ValidationError
from typing import List, Dict

class Row(BaseModel):
    x: str = Field(min_length=2)
    y: str = Field(min_length=2)
    z: int = Field(gt=0, le=300001)

def validate_df(df: pd.DataFrame) -> tuple[List[Row], Dict[int, Dict]]:
    valid_rows = []
    error_records = {}
    
    for idx, row_data in df.iterrows():
        # 映射DataFrame列名到Row模型字段
        model_input = {
            "x": row_data["xval"],
            "y": row_data["yval"],
            "z": row_data["zval"]
        }
        try:
            validated_row = Row(**model_input)
            valid_rows.append(validated_row)
        except ValidationError as e:
            # 提取详细错误信息
            error_details = []
            for err in e.errors():
                field = err["loc"][0]
                msg = err["msg"]
                invalid_value = model_input[field]
                error_details.append({
                    "字段": field,
                    "错误信息": msg,
                    "无效值": invalid_value
                })
            error_records[idx] = error_details
    
    return valid_rows, error_records

# 示例调用
if __name__ == "__main__":
    # 模拟测试DataFrame
    test_data = {
        "xval": ["ab", "c", "def", "gh"],
        "yval": ["ij", "klm", "n", "op"],
        "zval": [100, "200", 300002, 500]
    }
    df = pd.DataFrame(test_data)
    
    valid_rows, error_records = validate_df(df)
    
    # 输出有效行
    print("有效行：")
    for row in valid_rows:
        print(row.model_dump())
    
    # 输出错误记录
    print("\n错误记录：")
    for idx, errors in error_records.items():
        print(f"行索引 {idx}:")
        for err in errors:
            print(f"  - {err['字段']}: {err['无效值']}，错误：{err['错误信息']}")
    
    # 高效转回DataFrame
    valid_df = pd.DataFrame([row.model_dump() for row in valid_rows])
    print("\n验证后的DataFrame：")
    print(valid_df)

问题解决细节 #

1. 跳过错误行并继续运行 #

将Row实例化逻辑包裹在try-except块中，精准捕获Pydantic的ValidationError而非通用异常。当某行数据验证失败时，程序会进入异常分支记录错误，随后继续处理下一行，避免整体崩溃。

2. 记录详细错误回溯信息 #

从ValidationError的errors()方法中提取结构化错误数据：

• loc：定位错误对应的字段名
• msg：Pydantic生成的具体错误提示（如"字符串长度不足2"、"不是有效的整数"）
• 结合原始输入值，将这些信息整理成可读字典并关联行索引，方便后续定位修复数据源问题。

3. 高效转回DataFrame #

利用Pydantic模型的model_dump()方法（Pydantic v2版本，v1用dict()），将每个验证通过的Row实例转为字典，直接传入pd.DataFrame()构造新的DataFrame。这种方式无需手动拼接字段，既高效又能严格遵循模型的字段命名规范。

内容的提问来源于stack exchange，提问作者Amogh Bhosekar