棉花糖`validates_schema`用`pass_many = True`拒絕未知字段

Question

當Schema傳遞一個用於驗證的對象列表時，我很難理解如何處理未知字段。我走到這一步：

class MySchema(Schema): 
    # fields ... 

    @marshmallow_decorators.validates_schema(pass_original=True) 
    def check_unknown_fields(self, data, original_data): 
     if isinstance(original_data, list): 
      for dct in original_data: 
       self._assert_no_unknown_field(dct) 
     else: 
      self._assert_no_unknown_field(original_data) 

    def _assert_no_unknown_field(self, dct): 
     unknown = set(dct.keys()) - set(self.fields) 
     if unknown: 
      raise MarshmallowValidationError('Unknown field', unknown) 

但是，這顯然是行不通的，因爲驗證器運行列表中的每一次的所有項目。因此，第一個錯誤會被抓住，並在所有項目返回：

items = [ 
    {'a': 1, 'b': 2, 'unknown1': 3}, 
    {'a': 4, 'b': 5, 'unknown2': 6}, 
] 
errors = MySchema(many=True).validate(items) 
# {0: {'unknown1': ['Unknown field']}, 1: {'unknown1': ['Unknown field']}} 

我試圖想辦法來對應data說法從original_data只能得到單一的項目，只有驗證，但我不能真的這樣做，因爲物品沒有ID或字段，這將使他們可搜索...

我錯過了什麼嗎？有針對這個的解決方法嗎？

Answer 1

這是我想出了......一個解決辦法我希望它是簡單的，但在這裏它是：

from marshmallow import Schema, ValidationError as MarshmallowValidationError, fields 

UNKNOWN_MESSAGE = 'unknown field' 


class _RejectUnknownMixin(object): 

    def _collect_unknown_fields_errors(self, schema, data): 
     """ 
     Checks `data` against `schema` and returns a dictionary `{<field>: <error>}` 
     if unknown fields detected, or `{0: {<field>: <error>}, ... N: <field>: <error>}` 
     if `data` is a list. 
     """ 
     if isinstance(data, list): 
      validation_errors = {} 
      for i, datum in enumerate(data): 
       datum_validation_errors = self._collect_unknown_fields_errors(schema, datum) 
       if datum_validation_errors: 
        validation_errors[i] = datum_validation_errors 
      return validation_errors 

     else: 
      unknown = set(data.keys()) - set(schema.fields) 
      return {name: [UNKNOWN_MESSAGE] for name in unknown} 


class NestedRejectUnknown(fields.Nested, _RejectUnknownMixin): 
    """ 
    Nested field that returns validation errors if unknown fields are detected. 
    """ 

    def _deserialize(self, value, attr, data): 
     validation_errors = {} 
     try: 
      result = super(NestedRejectUnknown, self)._deserialize(value, attr, data) 
     except MarshmallowValidationError as err: 
      validation_errors = err.normalized_messages() 

     # Merge with unknown field errors 
     validation_errors = _merge_dicts(
      self._collect_unknown_fields_errors(self.schema, value), validation_errors) 
     if validation_errors: 
      raise MarshmallowValidationError(validation_errors) 

     return result 


class SchemaRejectUnknown(Schema, _RejectUnknownMixin): 
    """ 
    Schema that return validation errors if unknown fields are detected 
    """ 

    def validate(self, data, **kwargs): 
     validation_errors = super(SchemaRejectUnknown, self).validate(data, **kwargs) 
     return _merge_dicts(
      self._collect_unknown_fields_errors(self, data), validation_errors) 


def _merge_dicts(a, b, path=None): 
    """ 
    Ref : https://stackoverflow.com/questions/7204805/dictionaries-of-dictionaries-merge 
    merges b into a 
    """ 
    if path is None: 
     path = [] 
    for key in b: 
     if key in a: 
      if isinstance(a[key], dict) and isinstance(b[key], dict): 
       _merge_dicts(a[key], b[key], path + [str(key)]) 
      elif a[key] == b[key]: 
       # same leaf value 
       pass 
      else: 
       raise Exception('Conflict at %s' % '.'.join(path + [str(key)])) 
     else: 
      a[key] = b[key] 
    return a