如何用Pydantic验证“并行”格式JSON数组并实现灵活访问？

针对这种类似pandas split格式的JSON数据验证与转换需求，这里提供两种基于Pydantic的实现方案，解决你提到的三个痛点：

方案1：验证后转换为可按列名访问的结构 #

这种方案先完成基础验证，再通过Pydantic的后置验证器将data数组转换为带属性访问的对象列表，无需硬编码列顺序，同时支持可变列。

代码实现 #

from pydantic import BaseModel, model_validator
from typing import List, Dict, Any

class ValueRange(BaseModel):
    low: int
    high: int

# 定义所有可能返回的列名与对应类型的映射
COLUMN_TYPE_MAP = {
    "sensor": str,
    "value": int,
    "range": ValueRange,
}

class SplitResponse(BaseModel):
    columns: List[str]
    data: List[List[Any]]

    @model_validator(mode='after')
    def transform_data_to_rows(self) -> 'SplitResponse':
        # 验证列名是否合法（可选，按需开启）
        for col in self.columns:
            if col not in COLUMN_TYPE_MAP:
                raise ValueError(f"未知列名: {col}")
        
        # 根据当前列集合动态生成Row模型
        class Row(BaseModel):
            __annotations__ = {col: COLUMN_TYPE_MAP[col] for col in self.columns}
        
        # 将每行数据映射为Row实例
        self.data = [Row(**dict(zip(self.columns, row))) for row in self.data]
        return self

使用示例 #

# 带range列的响应
json_data1 = {
    "columns": ["sensor", "value", "range"],
    "data": [["a", 1, {"low": 1, "high": 2}], ["b", 2, {"low": 0, "high": 2}]]
}
response1 = SplitResponse(**json_data1)
print(response1.data[0].sensor)  # 输出: a
print(response1.data[0].range.low)  # 输出: 1

# 仅含sensor和value列的响应
json_data2 = {
    "columns": ["sensor", "value"],
    "data": [["a", 1], ["b", 2]]
}
response2 = SplitResponse(**json_data2)
print(response2.data[0].value)  # 输出: 1

方案2：动态生成模型做严格前置验证 #

如果需要在验证阶段就严格校验每列数据类型，而非转换阶段再处理，可以动态生成对应列的Row模型，结合前置验证器实现。

代码实现 #

from pydantic import BaseModel, model_validator, ValidationError
from typing import List, Type, Dict, Any

class ValueRange(BaseModel):
    low: int
    high: int

COLUMN_TYPE_MAP = {
    "sensor": str,
    "value": int,
    "range": ValueRange,
}

def create_row_model(columns: List[str]) -> Type[BaseModel]:
    """根据列名动态生成对应的Row验证模型"""
    annotations = {col: COLUMN_TYPE_MAP[col] for col in columns}
    return type("DynamicRow", (BaseModel,), {"__annotations__": annotations})

class SplitResponse(BaseModel):
    columns: List[str]
    data: List[List]

    @model_validator(mode='before')
    def validate_data_against_columns(cls, values: Dict[str, Any]) -> Dict[str, Any]:
        columns = values.get("columns")
        data = values.get("data")
        
        if not columns or not data:
            return values
        
        # 校验列名合法性
        for col in columns:
            if col not in COLUMN_TYPE_MAP:
                raise ValueError(f"无效列名: {col}")
        
        # 动态生成模型并逐行验证
        RowModel = create_row_model(columns)
        validated_data = []
        for idx, row in enumerate(data):
            if len(row) != len(columns):
                raise ValueError(f"第{idx+1}行数据长度与列数不匹配")
            try:
                validated_row = RowModel(**dict(zip(columns, row)))
                validated_data.append(validated_row)
            except ValidationError as e:
                raise ValueError(f"第{idx+1}行数据验证失败: {e}")
        
        values["data"] = validated_data
        return values

优势 #

• 在数据进入模型前就完成类型校验，提前拦截非法数据
• 转换后的data是带类型提示的模型实例，支持IDE自动补全

补充说明 #

• 如果不需要严格类型校验，仅需按列名访问，可直接将data转换为List[Dict[str, Any]]，用字典键值对访问（如response.data[0]['sensor']），实现更简洁。
• 若接口返回列名完全不可预测，可将列类型设为Any，但会失去类型校验能力。

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