Marshmallow中实现嵌套Schema的Union类型字段需求问询

问题描述 #

我希望实现如下代码逻辑，让Test Schema的attr5字段能够容纳两种不同的嵌套Schema（Foo或Bar）：

class Foo(Schema):
    attr1 = fds.Str(required=True)
    attr2 = fds.Int(required=True)

class Bar(Schema):
    attr3 = fds.Bool(required=True)
    attr4 = fds.Float(required=True)

class Test(Schema):
    attr5 = Union[Nested(Foo), Nested(Bar)]

但目前找到的Union实现仅支持普通数据类型，不支持嵌套Schema，请问这个需求是否可以实现？

解决方案 #

这个需求完全可以实现，以下是两种实用的实现方案：

方案1：自定义嵌套Union字段类 #

继承基础字段类，在反序列化时依次尝试用目标嵌套Schema解析数据，只要任意一个Schema验证通过即可：

from your_schema_lib import Schema, fields as fds

class Foo(Schema):
    attr1 = fds.Str(required=True)
    attr2 = fds.Int(required=True)

class Bar(Schema):
    attr3 = fds.Bool(required=True)
    attr4 = fds.Float(required=True)

class UnionNestedField(fds.Field):
    def __init__(self, schema_classes, **kwargs):
        # 初始化Schema实例
        self.schemas = [cls() for cls in schema_classes]
        super().__init__(**kwargs)
    
    def _deserialize(self, value, attr, data, **kwargs):
        validation_errors = []
        for schema in self.schemas:
            try:
                # 尝试用当前Schema加载数据
                return schema.load(value)
            except Exception as e:
                validation_errors.append(str(e))
        # 所有Schema都验证失败时抛出错误
        raise ValueError(f"数据不符合任何嵌套Schema要求: {'; '.join(validation_errors)}")

# 使用自定义字段
class Test(Schema):
    attr5 = UnionNestedField([Foo, Bar], required=True)

方案2：直接添加自定义验证函数 #

如果你的Schema库支持为字段指定验证函数，可以直接在attr5字段上绑定验证逻辑：

from your_schema_lib import Schema, fields as fds

class Foo(Schema):
    attr1 = fds.Str(required=True)
    attr2 = fds.Int(required=True)

class Bar(Schema):
    attr3 = fds.Bool(required=True)
    attr4 = fds.Float(required=True)

def validate_union_nested(value):
    foo_schema = Foo()
    bar_schema = Bar()
    # 先尝试用Foo解析
    try:
        return foo_schema.load(value)
    except:
        # 再尝试用Bar解析
        try:
            return bar_schema.load(value)
        except:
            raise ValueError("数据必须匹配Foo或Bar的Schema结构")

class Test(Schema):
    attr5 = fds.Field(validate=validate_union_nested, required=True)

说明 #

两种方案的核心逻辑一致：在加载数据时依次尝试用目标嵌套Schema解析attr5的值，只要其中一个Schema验证通过就返回解析后的对象；若所有Schema都验证失败，则抛出明确的错误提示。你可以根据使用的Schema库（如marshmallow、Pydantic等）的具体API调整细节，比如错误处理的方式、Schema实例化的逻辑等。

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