如何构建适配AWS SimpleDB接口的Marshmallow通用转换Schema

解决方案 #

你不需要拆分出单独的Nested Schema，直接自定义一个通用字段处理字典和Name/Value列表的双向转换即可，同时可以内置校验逻辑：

from marshmallow import fields, ValidationError, Schema

class SDBAttributesField(fields.Field):
    """
    双向转换Python字典和SimpleDB要求的Name/Value列表格式
    序列化：字典 -> [{"Name": k, "Value": v}, ...]
    反序列化：[{"Name": k, "Value": v}, ...] -> 字典
    内置校验：键值均为字符串类型
    """
    def _serialize(self, value, attr, obj, **kwargs):
        # 校验输入是字典，且键值都是字符串
        if not isinstance(value, dict):
            raise ValidationError("Attributes必须为字典类型")
        for k, v in value.items():
            if not isinstance(k, str):
                raise ValidationError(f"属性键{repr(k)}必须为字符串类型")
            if not isinstance(v, str):
                raise ValidationError(f"属性键{repr(k)}对应的值必须为字符串类型")
        # 转换为目标格式
        return [{"Name": k, "Value": v} for k, v in value.items()]
    
    def _deserialize(self, value, attr, data, **kwargs):
        # 校验输入是列表，每个元素符合格式要求
        if not isinstance(value, list):
            raise ValidationError("Attributes必须为列表类型")
        res = {}
        for item in value:
            if not isinstance(item, dict) or "Name" not in item or "Value" not in item:
                raise ValidationError(f"属性项{repr(item)}格式错误，必须包含Name和Value字段")
            k = item["Name"]
            v = item["Value"]
            if not isinstance(k, str):
                raise ValidationError(f"属性键{repr(k)}必须为字符串类型")
            if not isinstance(v, str):
                raise ValidationError(f"属性键{repr(k)}对应的值必须为字符串类型")
            res[k] = v
        return res

Schema直接调用该自定义字段即可：

class SDBItemSchema(Schema):
    Name = fields.Str(required=True)
    Attributes = SDBAttributesField(required=True)

测试用例可完全匹配你的预期：

# 测试序列化
record_to_dump = {'Name': 'test', 'Attributes': {'key1': 'value1', 'key2': 'value2'}}
schema = SDBItemSchema()
dumped = schema.dump(record_to_dump)
assert dumped == {
    "Name": "test",
    "Attributes": [{"Name": "key1", "Value" : "value1"}, {"Name": "key2", "Value": "value2"}]
}
# 测试反序列化
loaded = schema.load(dumped)
assert loaded == record_to_dump

困惑点解答 #

• 关于fields.Dict的序列化问题：默认的fields.Dict仅会原样序列化字典结构，不会自动转换为SimpleDB要求的Name/Value列表格式，因此不会匹配你预期的输出结构，需要自定义字段实现转换逻辑。
• 关于字符串校验：上面的自定义字段内置了全链路校验逻辑，序列化阶段会检查输入字典的键值是否均为字符串，反序列化阶段会检查返回的Name/Value列表的对应字段是否均为字符串，符合你的校验要求。

如果有动态字段类型的需求，比如部分属性是数字、日期类型需要用到你之前写的SDBNumberField等自定义编码字段，可以结合pre_dump和post_load钩子，先对字典的属性值做类型编码/解码，再走上面的转换逻辑即可。

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