Marshmallow `validates_schema`, um unbekannte Felder mit` pass_many = True` abzulehnen

Question

Ich habe Schwierigkeiten zu verstehen, wie mit unbekannten Feldern umzugehen ist, wenn die Schema eine Liste von Objekten zur Validierung übergeben wird. Ich habe bisher:

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) 

Aber das offensichtlich nicht funktioniert, da der Validator für alle Elemente in der Liste läuft jedes Mal. Daher ist der erste Fehler wird auf allen Artikeln gefangen und zurückgeschickt werden:

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']}} 

ich eine Art und Weise zu denken versuche nur das einzelne Element erhalten von original_data zum data Argumente entsprechen, und bestätigt nur, dass man, aber ich kann das nicht wirklich tun, da Elemente keine ID oder Feld haben, das sie durchsuchbar machen würde ...

Fehle ich etwas? Gibt es dafür eine Lösung?

Answer 1

Dies ist eine Abhilfe, die ich kam mit ... Ich wünschte, es wäre einfacher, aber hier ist es:

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