Marshmallow嵌套键加载失败求助:预加载处理器失效
解决Marshmallow加载嵌套字典键时的"Is not a valid string"错误
我来帮你分析下问题的根源,以及对应的解决方案:
问题原因
你的代码报错核心在于两个点:
pre_load装饰器可能未生效:如果代码里没正确导入marshmallow.pre_load,处理器不会运行,输入数据完全没被修改。此时load_from='brand__name'找不到对应键,Marshmallow会 fallback 到读取字段名brand的值——也就是输入里的字典,而String字段无法把字典转换成字符串,自然抛出"Is not a valid string"错误。- 即使处理器生效,逻辑也有冗余:你在
pre_load里新增了brand__name这类键,但原有的brand嵌套字典键仍然存在。虽然load_from会优先读取指定键,但这种新增键的方式不够直接,还可能留下潜在的逻辑冲突。
解决方案1:修复pre_load逻辑(最简单直接)
确保导入pre_load,然后修改处理器直接替换原嵌套键为对应的值,Schema字段无需load_from即可正确读取:
from marshmallow import Schema, fields, pre_load import unittest class ResultSchema(Schema): title = fields.String() description = fields.String() brand = fields.String() price = fields.String() url = fields.String() @pre_load def flatten_fields(self, data): # 定义字段与嵌套键的映射关系 field_mappings = [ ("brand", "name"), ("price", "value"), ("url", "value") ] for field, nested_key in field_mappings: try: # 直接把嵌套值替换到原字段位置 data[field] = data[field][nested_key] except (KeyError, TypeError): # 字段不存在或不是字典时,不做修改(也可设为None) pass return data class SchemaTests(unittest.TestCase): def setUp(self): self.resultSchema = ResultSchema() self.expected = { 'title': 'fake title', 'description': 'fake description', 'brand': {'name': 'fake brand name'}, 'price': {'value': '$82.99', 'integral': 8299}, 'url': {'value': 'fake url'}, 'images': [ {'value': 'small url', 'xsize': 60, 'ysize': 60}, {'value': 'small-medium url', 'xsize': 100, 'ysize': 100}, {'value': 'medium-large url', 'xsize': 160, 'ysize': 160}, {'value': 'large url', 'xsize': 400, 'ysize': 400} ] } def test_schema_load(self): loaded, err = self.resultSchema.load(self.expected) if err: self.fail(str(err)) # 可选:添加断言验证结果正确性 self.assertEqual(loaded['brand'], 'fake brand name') self.assertEqual(loaded['price'], '$82.99')
解决方案2:自定义嵌套路径字段(更灵活)
如果不想修改原始输入数据,可以自定义一个字段,专门从嵌套路径中提取值:
from marshmallow import Schema, fields, ValidationError import unittest class NestedPathField(fields.String): def __init__(self, nested_path, **kwargs): self.nested_path = nested_path.split('.') super().__init__(**kwargs) def _deserialize(self, value, attr, data, **kwargs): # 从完整输入数据中按路径逐层提取值 current_data = data for key in self.nested_path: if not isinstance(current_data, dict) or key not in current_data: # 路径不存在时返回None,也可抛出ValidationError return None current_data = current_data[key] # 调用父类逻辑确保值是合法字符串 return super()._deserialize(current_data, attr, data, **kwargs) class ResultSchema(Schema): title = fields.String() description = fields.String() brand = NestedPathField(nested_path='brand.name') price = NestedPathField(nested_path='price.value') url = NestedPathField(nested_path='url.value') # 测试代码与之前一致 class SchemaTests(unittest.TestCase): def setUp(self): self.resultSchema = ResultSchema() self.expected = { 'title': 'fake title', 'description': 'fake description', 'brand': {'name': 'fake brand name'}, 'price': {'value': '$82.99', 'integral': 8299}, 'url': {'value': 'fake url'}, 'images': [ {'value': 'small url', 'xsize': 60, 'ysize': 60}, {'value': 'small-medium url', 'xsize': 100, 'ysize': 100}, {'value': 'medium-large url', 'xsize': 160, 'ysize': 160}, {'value': 'large url', 'xsize': 400, 'ysize': 400} ] } def test_schema_load(self): loaded, err = self.resultSchema.load(self.expected) if err: self.fail(str(err)) self.assertEqual(loaded['brand'], 'fake brand name')
内容的提问来源于stack exchange,提问作者Christopher Engelbrecht
相关产品推荐
相关产品推荐

