Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __validate_logical(self, operator, definitions, field, value):
""" Validates value against all definitions and logs errors according
to the operator. """
valid_counter = 0
_errors = errors.ErrorList()
for i, definition in enumerate(definitions):
schema = {field: definition.copy()}
for rule in ('allow_unknown', 'type'):
if rule not in schema[field] and rule in self.schema[field]:
schema[field][rule] = self.schema[field][rule]
if 'allow_unknown' not in schema[field]:
schema[field]['allow_unknown'] = self.allow_unknown
validator = self._get_child_validator(
schema_crumb=(field, operator, i), schema=schema, allow_unknown=True
)
if validator(self.document, update=self.update, normalize=False):
valid_counter += 1
else:
self._drop_nodes_from_errorpaths(validator._errors, [], [3])
def fetch_errors_from(self, path: DocumentPath) -> ErrorList:
""" Returns all errors for a particular path. """
node = self.fetch_node_from(path)
if node is None:
return ErrorList()
else:
return node.errors
def __validate_logical(self, operator, definitions, field, value):
""" Validates value against all definitions and logs errors according
to the operator. """
valid_counter = 0
_errors = errors.ErrorList()
for i, definition in enumerate(definitions):
schema = {field: definition.copy()}
for rule in ('allow_unknown', 'type'):
if rule not in schema[field] and rule in self.schema[field]:
schema[field][rule] = self.schema[field][rule]
if 'allow_unknown' not in schema[field]:
schema[field]['allow_unknown'] = self.allow_unknown
validator = self._get_child_validator(
schema_crumb=(field, operator, i), schema=schema, allow_unknown=True
)
if validator(self.document, update=self.update, normalize=False):
valid_counter += 1
else:
self._drop_nodes_from_errorpaths(validator._errors, [], [3])
def __init_processing(self, document, schema=None):
self._errors = errors.ErrorList()
self.recent_error = None
self.document_error_tree = errors.DocumentErrorTree()
self.schema_error_tree = errors.SchemaErrorTree()
self.document = copy(document)
if not self.is_child:
self._is_normalized = False
if schema is not None:
self.schema = DefinitionSchema(self, schema)
elif self.schema is None:
if isinstance(self.allow_unknown, Mapping):
self._schema = {}
else:
raise SchemaError(errors.SCHEMA_ERROR_MISSING)
if document is None:
raise DocumentError(errors.DOCUMENT_MISSING)
def _validate_anyof(self, definitions, field, value):
""" {'type': 'list', 'logical': 'anyof'} """
if 'role' not in list(definitions[0]['schema'].keys()):
return super(AnyofValidator, self)._validate_anyof(definitions,
field, value)
valids = 0
found_valid_role = False
_errors = errors.ErrorList()
valid_roles = []
for i, definition in enumerate(definitions):
valid_roles.extend(definition['schema']['role']['allowed'])
if value['role'] not in definition['schema']['role']['allowed']:
continue
found_valid_role = True
schema = {field: definition.copy()}
for rule in ('allow_unknown', 'type'):
if rule not in schema[field] and rule in self.schema[field]:
schema[field][rule] = self.schema[field][rule]
if 'allow_unknown' not in schema[field]:
schema[field]['allow_unknown'] = self.allow_unknown
validator = self._get_child_validator(
def __init_processing(self, document, schema=None):
self._errors = errors.ErrorList()
self.recent_error = None
self.document_error_tree = errors.DocumentErrorTree()
self.schema_error_tree = errors.SchemaErrorTree()
self.document = copy(document)
if not self.is_child:
self._is_normalized = False
self.__init_schema(schema)
if self.schema is None:
if isinstance(self.allow_unknown, Mapping):
self.schema = {}
else:
raise SchemaError(errors.MISSING_SCHEMA)
if document is None:
def __init__(self, *args, **kwargs):
""" The arguments will be treated as with this signature:
__init__(self, schema=None, ignore_none_values=False,
allow_unknown=False, require_all=False,
purge_unknown=False, purge_readonly=False,
error_handler=errors.BasicErrorHandler)
"""
self.document = None
""" The document that is or was recently processed.
Type: any :term:`mapping` """
self._errors = errors.ErrorList()
""" The list of errors that were encountered since the last document
processing was invoked.
Type: :class:`~cerberus.errors.ErrorList` """
self.recent_error = None
""" The last individual error that was submitted.
Type: :class:`~cerberus.errors.ValidationError` """
self.document_error_tree = errors.DocumentErrorTree()
""" A tree representiation of encountered errors following the
structure of the document.
Type: :class:`~cerberus.errors.DocumentErrorTree` """
self.schema_error_tree = errors.SchemaErrorTree()
""" A tree representiation of encountered errors following the
structure of the schema.
Type: :class:`~cerberus.errors.SchemaErrorTree` """
self.document_path = ()
""" The path within the document to the current sub-document.