如何用Pandera验证数据类型不匹配并抛错（动态Schema场景）

我需要实现两个需求：

• 使用Pandera验证非数值列与Schema中定义的数值类型不匹配时抛出错误
• 由于Schema从用户配置文件动态生成，需了解如何以编程方式通过DataFrameModel定义Schema

当前我定义的Schema代码如下：

import re
from typing import Any, Dict, List, Pattern

import pandas as pd
import numpy as np
import pandera as pa

pa_schema_dict: Dict[str, pa.Column] = {
    item[0]: pa.Column(
        dtype=item[1],
        nullable=True,
        required=True,
        name=item[0],
        drop_invalid_rows=True,
    )
    for item in [("housing_median_age", "int"), ("ocean_proximity", "int")]
}

for name, value in pa_schema_dict.items():
    if value.dtype.type in ("float", "int"):
        print(f"{name}: {value.dtype.type}")
        # check: pa.Check = pa.Check(
        #     lambda x: (
        #         True
        #         if re.match(
        #             pattern=r"^-?\d+(\.\d+)?([eE]-?\d+)?$",
        #             string=str(x),
        #             flags=re.ASCII,
        #         )
        #         is not None
        #         else False
        #     ),
        # )
        check: pa.Check = pa.Check(
            pd.api.types.is_numeric_dtype,
        )
        if value.checks:
            value.checks.append(check)
        else:
            value.checks = [check]

pa_feature_schema: pa.DataFrameSchema = pa.DataFrameSchema(
    columns=pa_schema_dict,
    name="data_schema",
    drop_invalid_rows=True,
    coerce=True,
    strict="filter",
    add_missing_columns=False,
)

对应的DataFrame：

df: pd.DataFrame = pd.DataFrame(
    {"housing_median_age": [6, 25, np.nan, 25],
     "ocean_proximity": ["NEAR BAY", "INLAND", "<1H OCEAN", "NEAR OCEAN"]
     })

其中Schema中ocean_proximity定义为int64类型，但DataFrame中该列是object类型，调用以下验证代码时未抛出预期异常：

try:
    validated_df: pd.DataFrame = pa_feature_schema.validate(
        check_obj=df, lazy=True, inplace=True
    )
except (pa.errors.SchemaError, pa.errors.SchemaDefinitionError,) as e:
    print(str(e))

一、修复非数值列类型不匹配未报错的问题 #

你的代码存在两个核心问题导致验证失效：

1. coerce=True的干扰：该参数会让Pandera尝试强制转换列类型到Schema定义的类型，转换失败时，drop_invalid_rows=True会直接丢弃无效行而非抛出错误。要触发错误，需关闭强制转换并禁用无效行丢弃。
2. 错误的Check逻辑：pd.api.types.is_numeric_dtype用于检查列的整体数据类型，而非元素级验证，且未设置element_wise=True，无法生效。实际上无需额外添加Check，Pandera本身的dtype验证足以检测类型不匹配。

修改后的Schema代码：

import pandas as pd
import numpy as np
import pandera as pa
from typing import Dict

# 定义Schema字典
pa_schema_dict: Dict[str, pa.Column] = {
    item[0]: pa.Column(
        dtype=item[1],
        nullable=True,
        required=True,
        name=item[0],
        drop_invalid_rows=False,  # 不丢弃无效行，触发错误
    )
    for item in [("housing_median_age", "int"), ("ocean_proximity", "int")]
}

pa_feature_schema: pa.DataFrameSchema = pa.DataFrameSchema(
    columns=pa_schema_dict,
    name="data_schema",
    coerce=False,  # 关闭强制类型转换，类型不匹配直接报错
    strict="filter",
    add_missing_columns=False,
)

运行原验证代码，会抛出SchemaError，提示ocean_proximity列的object类型与预期的int64不匹配。

若需更严格的元素级验证（比如禁止object列中存在可转成数值的字符串），可给数值列添加元素级Check：

for name, col in pa_schema_dict.items():
    if col.dtype in ("int", "float"):
        # 检查每个元素是否为数值或空值
        check = pa.Check(lambda x: pd.api.types.is_number(x) or pd.isna(x), element_wise=True)
        col.checks.append(check)

二、动态生成DataFrameModel #

可通过Python的type()函数动态构造DataFrameModel，核心是根据配置生成模型字段字典，再创建类。

示例代码：

import pandas as pd
import numpy as np
import pandera as pa
from pandera.typing import Series

# 模拟从配置文件读取的字段定义：(列名, 类型, 是否可空)
config_fields = [
    ("housing_median_age", "int", True),
    ("ocean_proximity", "int", True)
]

# 构造模型字段字典
model_fields = {}
for name, dtype_str, nullable in config_fields:
    # 映射字符串类型到Pandera类型注解
    if dtype_str == "int":
        dtype = Series[int]
    elif dtype_str == "float":
        dtype = Series[float]
    elif dtype_str == "str":
        dtype = Series[str]
    else:
        raise ValueError(f"不支持的类型: {dtype_str}")
    
    # 添加字段约束
    field_kwargs = {"nullable": nullable}
    # 给数值列添加元素级检查
    if dtype_str in ("int", "float"):
        field_kwargs["checks"] = [pa.Check(lambda x: pd.api.types.is_number(x) or pd.isna(x), element_wise=True)]
    
    model_fields[name] = dtype.field(**field_kwargs)

# 动态生成DataFrameModel类
DynamicSchema = type(
    "DynamicSchema",  # 模型名称
    (pa.DataFrameModel,),  # 继承自DataFrameModel
    model_fields  # 字段定义
)

# 使用动态Schema验证DataFrame
try:
    validated_df = DynamicSchema.validate(df, lazy=True)
except pa.errors.SchemaError as e:
    print(str(e))

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