将Marshmallow Schema重构为Pydantic 2时的动态AttributeValue字段实现问题

我之前也碰到过类似的动态字段依赖场景，在Pydantic 2里可以通过模型验证器+FieldValidationInfo完美复刻你原来Marshmallow的逻辑。核心思路是利用Pydantic的上下文传递机制获取manifest，再结合同模型的id字段动态处理value的类型转换和验证。下面是完整的重构方案：

一、核心逻辑对齐 #

先明确要复刻的Marshmallow关键行为，确保重构后业务逻辑完全一致：

1. 根据attribute的id从manifest中匹配属性类型（single/multi/boolean等）
2. 按类型自动转换value格式（比如single转字符串、multi转字符串列表）
3. 若属性配置了option_labels，则按类型执行单选/多选验证
4. 禁止value为null，同时处理manifest中不存在的attribute id场景

二、Pydantic 2实现代码 #

1. 适配Pydantic的验证函数

把原来的验证器调整为Pydantic兼容格式，直接抛出ValueError或ValidationError即可，Pydantic会自动包装成标准错误结构：

from pydantic import BaseModel, ValidationError, field_validator, FieldValidationInfo
from typing import Union, list, str, bool, int

def make_single_option_validator(options: list[str]):
    def validate(value: str) -> str:
        if value not in options:
            raise ValueError(f'"{value}" is not a valid option')
        return value
    return validate

def make_multiple_option_validator(options: list[str]):
    def validate(values: list[str]) -> list[str]:
        invalid_values = set(values) - set(options)
        if invalid_values:
            invalid = ', '.join(sorted(f'"{v}"' for v in invalid_values))
            raise ValueError(f"The following values are not valid options: {invalid}")
        return values
    return validate

2. 定义Attribute Pydantic模型

将原来的dataclass转成Pydantic BaseModel，通过两个字段验证器实现动态逻辑：

class Attribute(BaseModel):
    id: str
    value: Union[str, list[str], bool, int]

    model_config = {
        "extra": "forbid",  # 禁止额外字段，和Marshmallow Schema行为一致
        "arbitrary_types_allowed": False
    }

    @field_validator('value', mode='before')
    @classmethod
    def validate_dynamic_value(cls, input_value: object, info: FieldValidationInfo) -> Union[str, list[str], bool, int, None]:
        # 从上下文获取manifest（对应Marshmallow的self.parent.manifest）
        manifest = info.context.get('manifest')
        if not manifest:
            raise ValidationError("Manifest must be provided in model context")
        
        # 获取同模型的id字段值（原始输入，mode='before'确保在value验证前拿到id）
        attribute_id = info.data.get('id')
        if not attribute_id:
            raise ValidationError("Attribute 'id' is required to validate 'value'")
        
        # 从manifest获取属性配置
        manifest_attr = manifest.attributes.get(attribute_id)
        if not manifest_attr:
            # 对应原来的null_field，返回None后由后续验证器报错
            return None
        
        attr_type = manifest_attr['type']
        
        # 类型转换逻辑（对应原来的type_fields）
        type_converters = {
            "single": lambda v: str(v),
            "multi": lambda v: list(v) if isinstance(v, (list, tuple)) else [str(v)],
            "boolean": lambda v: bool(v),
            "range": lambda v: int(v),
            "account": lambda v: str(v),
            "supplier": lambda v: list(v) if isinstance(v, (list, tuple)) else [str(v)],
        }

        try:
            converted_value = type_converters[attr_type](input_value)
        except KeyError:
            raise ValidationError(f"Unknown attribute type '{attr_type}' for id '{attribute_id}'")
        except (ValueError, TypeError):
            raise ValidationError(f"Value '{input_value}' cannot be converted to type '{attr_type}' for attribute '{attribute_id}'")
        
        # 选项验证逻辑（对应原来的option_validator_factories）
        skip_validation = info.context.get('skip_attributes_validation', False)
        if not skip_validation and attr_type in ["single", "account", "multi", "supplier"] and "option_labels" in manifest_attr:
            options = manifest_attr["option_labels"]
            if attr_type in ["single", "account"]:
                converted_value = make_single_option_validator(options)(converted_value)
            elif attr_type in ["multi", "supplier"]:
                converted_value = make_multiple_option_validator(options)(converted_value)
        
        return converted_value

    @field_validator('value', mode='after')
    @classmethod
    def ensure_value_not_null(cls, value: Union[str, list[str], bool, int, None], info: FieldValidationInfo) -> Union[str, list[str], bool, int]:
        # 对应原来的validates_schema：禁止value为null
        if value is None:
            attribute_id = info.data.get('id', 'unknown')
            raise ValidationError(f"{attribute_id!r} value may not be null")
        return value

3. 调整测试用例适配Pydantic

原来的测试只需把schema.load()改成Pydantic的model_validate，并通过context传递manifest：

from datetime import date
import pytest
from pydantic import ValidationError

# 沿用你原来的测试用例结构
TYPE_INPUTS = [
    # Single
    (
        {"id": "industry", "value": "appliances"},
        Attribute(id="industry", value="appliances"),
    ),
    # Multi
    (
        {"id": "day_part", "value": ["am", "pm"]},
        Attribute(id="day_part", value=["am", "pm"]),
    ),
    # Boolean
    ({"id": "restricted", "value": True}, Attribute(id="restricted", value=True)),
]

@pytest.fixture
def mock_manifest():
    # 模拟你的manifest对象
    class MockManifest:
        def __init__(self):
            self.attributes = {
                "industry": {"type": "single", "option_labels": ["appliances", "electronics"]},
                "day_part": {"type": "multi", "option_labels": ["am", "pm", "night"]},
                "restricted": {"type": "boolean"},
            }
    return MockManifest()

@pytest.mark.parametrize("payload, expected_attribute", TYPE_INPUTS)
def test_deserialization_succeeds(mock_manifest, payload, expected_attribute):
    attribute = Attribute.model_validate(payload, context={"manifest": mock_manifest})
    assert attribute == expected_attribute

三、关键细节说明 #

1. 上下文传递：通过model_validate的context参数传递manifest和skip_attributes_validation，完全对应Marshmallow的Schema上下文机制。
2. 动态字段依赖：利用FieldValidationInfo.data获取同模型的id字段值，实现和Marshmallow中data["id"]一样的逻辑。
3. 类型与验证分离：用两个field_validator分别处理动态类型转换/验证和空值检查，代码结构更清晰，对应原来的AttributeValue字段逻辑和validates_schema钩子。
4. 行为一致性：完全复刻原来的错误提示、类型转换规则和验证逻辑，确保迁移后业务行为无变化。

备注：内容来源于stack exchange，提问作者Sharmiko