marshmallow `validates_schema` pour rejeter les champs inconnus avec` pass_many = True`

Question

J'ai du mal à comprendre comment gérer les champs inconnus lorsque le Schema reçoit une liste d'objets pour validation. Je suis arrivé à ce jour:

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) 

Mais cela ne fonctionne évidemment pas, comme le validateur est couru pour tous les éléments de la liste à chaque fois. Par conséquent, la première erreur sera pris et retourné sur tous les points:

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

je tentais de penser à un moyen d'obtenir que le seul élément de original_data correspondant à l'argument data et valider que celui-là, mais je ne peut pas vraiment faire cela, car les articles n'ont pas d'identifiant, ou un champ qui les rendrait consultables ...

Ai-je raté quelque chose? Y a-t-il une solution à cela?

Answer 1

Ceci est une solution de contournement je suis venu avec ... Je souhaite qu'il était plus simple, mais la voici:

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