python-validictory 0.8.3-2 / usr / share / pyshared / validictory / validator.py

import re
import sys
import copy
import socket
from datetime import datetime
import warnings
from collections import Mapping, Container

if sys.version_info[0] == 3:
    _str_type = str
    _int_types = (int,)
else:
    _str_type = basestring
    _int_types = (int, long)


class SchemaError(ValueError):
    """
    errors encountered in processing a schema (subclass of :class:`ValueError`)
    """


class ValidationError(ValueError):
    """
    validation errors encountered during validation (subclass of
    :class:`ValueError`)
    """


def _generate_datetime_validator(format_option, dateformat_string):
    def validate_format_datetime(validator, fieldname, value, format_option):
        try:
            datetime.strptime(value, dateformat_string)
        except ValueError:
            raise ValidationError(
                "Value %(value)r of field '%(fieldname)s' is not in "
                "'%(format_option)s' format" % locals())

    return validate_format_datetime

validate_format_date_time = _generate_datetime_validator('date-time',
                                                         '%Y-%m-%dT%H:%M:%SZ')
validate_format_date = _generate_datetime_validator('date', '%Y-%m-%d')
validate_format_time = _generate_datetime_validator('time', '%H:%M:%S')


def validate_format_utc_millisec(validator, fieldname, value, format_option):
    if not isinstance(value, (int, float)):
        raise ValidationError("Value %(value)r of field '%(fieldname)s' is "
                              "not a number" % locals())

    if not value > 0:
        raise ValidationError("Value %(value)r of field '%(fieldname)s' is "
                              "not a positive number" % locals())


def validate_format_ip_address(validator, fieldname, value, format_option):
    try:
        socket.inet_aton(value)
        # Make sure we expect "X.X.X.X" as socket.inet_aton() converts "1"
        # to "0.0.0.1"
        ip = len(value.split('.')) == 4
    except:
        ip = False
    if not ip:
        raise ValidationError("Value %(value)r of field '%(fieldname)s' is "
                              "not a ip-address" % locals())


DEFAULT_FORMAT_VALIDATORS = {
    'date-time': validate_format_date_time,
    'date': validate_format_date,
    'time': validate_format_time,
    'utc-millisec': validate_format_utc_millisec,
    'ip-address': validate_format_ip_address,
}


class SchemaValidator(object):
    '''
    Validator largely based upon the JSON Schema proposal but useful for
    validating arbitrary python data structures.

    :param format_validators: optional dictionary of custom format validators
    :param required_by_default: defaults to True, set to False to make
        ``required`` schema attribute False by default.
    :param blank_by_default: defaults to False, set to True to make ``blank``
        schema attribute True by default.
    '''

    def __init__(self, format_validators=None, required_by_default=True,
                 blank_by_default=False):
        if format_validators is None:
            format_validators = DEFAULT_FORMAT_VALIDATORS.copy()

        self._format_validators = format_validators
        self.required_by_default = required_by_default
        self.blank_by_default = blank_by_default

    def register_format_validator(self, format_name, format_validator_fun):
        self._format_validators[format_name] = format_validator_fun

    def validate_type_string(self, val):
        return isinstance(val, _str_type)

    def validate_type_integer(self, val):
        return type(val) in _int_types

    def validate_type_number(self, val):
        return type(val) in _int_types + (float,)

    def validate_type_boolean(self, val):
        return type(val) == bool

    def validate_type_object(self, val):
        return isinstance(val, Mapping)

    def validate_type_array(self, val):
        return isinstance(val, (list, tuple))

    def validate_type_null(self, val):
        return val is None

    def validate_type_any(self, val):
        return True

    def _error(self, desc, value, fieldname, **params):
        params['value'] = value
        params['fieldname'] = fieldname
        message = desc % params
        raise ValidationError(message)

    def validate_type(self, x, fieldname, schema, fieldtype=None):
        '''
        Validates that the fieldtype specified is correct for the given
        data
        '''

        # We need to know if the field exists or if it's just Null
        fieldexists = True
        try:
            value = x[fieldname]
        except KeyError:
            fieldexists = False
            value = None

        if fieldtype and fieldexists:
            if isinstance(fieldtype, (list, tuple)):
                # Match if type matches any one of the types in the list
                datavalid = False
                for eachtype in fieldtype:
                    try:
                        self.validate_type(x, fieldname, eachtype, eachtype)
                        datavalid = True
                        break
                    except ValidationError:
                        pass
                if not datavalid:
                    self._error("Value %(value)r for field '%(fieldname)s' is "
                                "not of type %(fieldtype)s",
                                value, fieldname, fieldtype=fieldtype)
            elif isinstance(fieldtype, dict):
                try:
                    self.__validate(fieldname, x, fieldtype)
                except ValueError as e:
                    raise e
            else:
                try:
                    type_checker = getattr(self, 'validate_type_%s' %
                                           fieldtype)
                except AttributeError:
                    raise SchemaError("Field type '%s' is not supported." %
                                      fieldtype)

                if not type_checker(value):
                    self._error("Value %(value)r for field '%(fieldname)s' "
                                "is not of type %(fieldtype)s",
                                value, fieldname, fieldtype=fieldtype)

    def validate_properties(self, x, fieldname, schema, properties=None):
        '''
        Validates properties of a JSON object by processing the object's
        schema recursively
        '''
        if x.get(fieldname) is not None:
            value = x.get(fieldname)
            if isinstance(value, dict):
                if isinstance(properties, dict):
                    for eachProp in properties:
                        self.__validate(eachProp, value,
                                        properties.get(eachProp))
                else:
                    raise SchemaError("Properties definition of field '%s' is "
                                      "not an object" % fieldname)

    def validate_items(self, x, fieldname, schema, items=None):
        '''
        Validates that all items in the list for the given field match the
        given schema
        '''
        if x.get(fieldname) is not None:
            value = x.get(fieldname)
            if isinstance(value, (list, tuple)):
                if isinstance(items, (list, tuple)):
                    if (not 'additionalItems' in schema and
                        len(items) != len(value)):
                        self._error("Length of list %(value)r for field "
                                    "'%(fieldname)s' is not equal to length "
                                    "of schema list", value, fieldname)
                    else:
                        for itemIndex in range(len(items)):
                            try:
                                self.validate(value[itemIndex],
                                              items[itemIndex])
                            except ValueError as e:
                                raise type(e)("Failed to validate field '%s' "
                                              "list schema: %s" %
                                              (fieldname, e))
                elif isinstance(items, dict):
                    for eachItem in value:
                        try:
                            self._validate(eachItem, items)
                        except ValueError as e:
                            # a bit of a hack: replace reference to _data
                            # with 'list item' so error messages make sense
                            old_error = str(e).replace("field '_data'",
                                                       'list item')
                            raise type(e)("Failed to validate field '%s' list "
                                          "schema: %s" %
                                          (fieldname, old_error))
                else:
                    raise SchemaError("Properties definition of field '%s' is "
                                      "not a list or an object" % fieldname)

    def validate_required(self, x, fieldname, schema, required):
        '''
        Validates that the given field is present if required is True
        '''
        # Make sure the field is present
        if fieldname not in x and required:
            self._error("Required field '%(fieldname)s' is missing",
                        None, fieldname)

    def validate_blank(self, x, fieldname, schema, blank=False):
        '''
        Validates that the given field is not blank if blank=False
        '''
        value = x.get(fieldname)
        if isinstance(value, _str_type) and not blank and not value:
            self._error("Value %(value)r for field '%(fieldname)s' cannot be "
                        "blank'", value, fieldname)

    def validate_patternProperties(self, x, fieldname, schema,
                                   patternproperties=None):

        if patternproperties == None:
            patternproperties = {}

        value_obj = x.get(fieldname, {})

        for pattern, schema in patternproperties.items():
            for key, value in value_obj.items():
                if re.match(pattern, key):
                    self.validate(value, schema)

    def validate_additionalItems(self, x, fieldname, schema,
                                 additionalItems=False):
        value = x.get(fieldname)

        if not isinstance(value, (list, tuple)):
            return

        if isinstance(additionalItems, bool):
            if additionalItems or 'items' not in schema:
                return
            elif len(value) != len(schema['items']):
                #print locals(), value, len(value), len(schema['items'])
                self._error("Length of list %(value)r for field "
                            "'%(fieldname)s' is not equal to length of schema "
                            "list", value, fieldname)

        remaining = value[len(schema['items']):]
        if len(remaining) > 0:
            self._validate(remaining, {'items': additionalItems})

    def validate_additionalProperties(self, x, fieldname, schema,
                                      additionalProperties=None):
        '''
        Validates additional properties of a JSON object that were not
        specifically defined by the properties property
        '''

        # Shouldn't be validating additionalProperties on non-dicts
        value = x.get(fieldname)
        if not isinstance(value, dict):
            return

        # If additionalProperties is the boolean value True then we accept
        # any additional properties.
        if isinstance(additionalProperties, bool) and additionalProperties:
            return

        value = x.get(fieldname)
        if isinstance(additionalProperties, (dict, bool)):
            properties = schema.get("properties")
            if properties is None:
                properties = {}
            if value is None:
                value = {}
            for eachProperty in value:
                if eachProperty not in properties:
                    # If additionalProperties is the boolean value False
                    # then we don't accept any additional properties.
                    if (isinstance(additionalProperties, bool) and
                        not additionalProperties):
                        self._error("additional properties not defined by "
                                    "'properties' are not allowed in field "
                                    "'%(fieldname)s'",
                                    None, fieldname)
                    self.__validate(eachProperty, value,
                                    additionalProperties)
        else:
            raise SchemaError("additionalProperties schema definition for "
                              "field '%s' is not an object" % fieldname)

    def validate_requires(self, x, fieldname, schema, requires=None):
        warnings.warn('The "requires" attribute has been replaced by "dependencies"', DeprecationWarning)
        if x.get(fieldname) is not None:
            if x.get(requires) is None:
                self._error("Field '%(requires)s' is required by field '%(fieldname)s'",
                            None, fieldname, requires=requires)

    def validate_dependencies(self, x, fieldname, schema, dependencies=None):
        if x.get(fieldname) is not None:

            # handle cases where dependencies is a string or list of strings
            if isinstance(dependencies, _str_type):
                dependencies = [dependencies]
            if isinstance(dependencies, (list, tuple)):
                for dependency in dependencies:
                    if dependency not in x:
                        self._error("Field '%(dependency)s' is required by "
                                    "field '%(fieldname)s'",
                            None, fieldname, dependency=dependency)
            elif isinstance(dependencies, dict):
                # NOTE: the version 3 spec is really unclear on what this means
                # based on the meta-schema I'm assuming that it should check
                # that if a key exists, the appropriate value exists
                for k, v in dependencies.items():
                    if k in x and v not in x:
                        self._error("Field '%(v)s' is required by field "
                                    "'%(k)s'", None, fieldname, k=k, v=v)
            else:
                raise SchemaError("'dependencies' must be a string, "
                                  "list of strings, or dict")

    def validate_minimum(self, x, fieldname, schema, minimum=None):
        '''
        Validates that the field is longer than or equal to the minimum
        length if specified
        '''

        exclusive = schema.get('exclusiveMinimum', False)

        if x.get(fieldname) is not None:
            value = x.get(fieldname)
            if value is not None:
                if (type(value) in (int, float) and
                    (not exclusive and value < minimum) or
                    (exclusive and value <= minimum)):
                    self._error("Value %(value)r for field '%(fieldname)s' is "
                                "less than minimum value: %(minimum)f",
                                value, fieldname, minimum=minimum)

    def validate_maximum(self, x, fieldname, schema, maximum=None):
        '''
        Validates that the field is shorter than or equal to the maximum
        length if specified.
        '''

        exclusive = schema.get('exclusiveMaximum', False)

        if x.get(fieldname) is not None:
            value = x.get(fieldname)
            if value is not None:
                if (type(value) in (int, float) and
                    (not exclusive and value > maximum) or
                    (exclusive and value >= maximum)):
                    self._error("Value %(value)r for field '%(fieldname)s' is "
                                "greater than maximum value: %(maximum)f",
                                value, fieldname, maximum=maximum)

    def validate_maxLength(self, x, fieldname, schema, length=None):
        '''
        Validates that the value of the given field is shorter than or equal
        to the specified length
        '''
        value = x.get(fieldname)
        if isinstance(value, (_str_type, list, tuple)) and len(value) > length:
            self._error("Length of value %(value)r for field '%(fieldname)s' "
                        "must be less than or equal to %(length)d",
                        value, fieldname, length=length)

    def validate_minLength(self, x, fieldname, schema, length=None):
        '''
        Validates that the value of the given field is longer than or equal
        to the specified length
        '''
        value = x.get(fieldname)
        if isinstance(value, (_str_type, list, tuple)) and len(value) < length:
            self._error("Length of value %(value)r for field '%(fieldname)s' "
                        "must be greater than or equal to %(length)d",
                        value, fieldname, length=length)

    validate_minItems = validate_minLength
    validate_maxItems = validate_maxLength

    def validate_format(self, x, fieldname, schema, format_option=None):
        '''
        Validates the format of primitive data types
        '''
        value = x.get(fieldname)

        format_validator = self._format_validators.get(format_option, None)

        if format_validator and value:
            format_validator(self, fieldname, value, format_option)

        # TODO: warn about unsupported format ?

    def validate_pattern(self, x, fieldname, schema, pattern=None):
        '''
        Validates that the given field, if a string, matches the given
        regular expression.
        '''
        value = x.get(fieldname)
        if isinstance(value, _str_type):
            if not re.match(pattern, value):
                self._error("Value %(value)r for field '%(fieldname)s' does "
                            "not match regular expression '%(pattern)s'",
                            value, fieldname, pattern=pattern)

    def validate_uniqueItems(self, x, fieldname, schema, uniqueItems=False):
        '''
        Validates that all items in an array instance MUST be unique
        (contains no two identical values).
        '''

        # If additionalProperties is the boolean value True then we accept
        # any additional properties.
        if isinstance(uniqueItems, bool) and not uniqueItems:
            return

        values = x.get(fieldname)

        if not isinstance(values, (list, tuple)):
            return

        hashables = set()
        unhashables = []

        for value in values:
            if isinstance(value, (list, dict)):
                container, add = unhashables, unhashables.append
            else:
                container, add = hashables, hashables.add

            if value in container:
                self._error(
                    "Value %(value)r for field '%(fieldname)s' is not unique",
                    value, fieldname)
            else:
                add(value)

    def validate_enum(self, x, fieldname, schema, options=None):
        '''
        Validates that the value of the field is equal to one of the
        specified option values
        '''
        value = x.get(fieldname)
        if value is not None:
            if not isinstance(options, Container):
                raise SchemaError("Enumeration %r for field '%s' must be a "
                                  "container", (options, fieldname))
            if value not in options:
                self._error("Value %(value)r for field '%(fieldname)s' is not "
                            "in the enumeration: %(options)r",
                            value, fieldname, options=options)

    def validate_title(self, x, fieldname, schema, title=None):
        if not isinstance(title, (_str_type, type(None))):
            raise SchemaError("The title for field '%s' must be a string" %
                             fieldname)

    def validate_description(self, x, fieldname, schema, description=None):
        if not isinstance(description, (_str_type, type(None))):
            raise SchemaError("The description for field '%s' must be a string"
                             % fieldname)

    def validate_divisibleBy(self, x, fieldname, schema, divisibleBy=None):
        value = x.get(fieldname)

        if not self.validate_type_number(value):
            return

        if divisibleBy == 0:
            raise SchemaError("'%r' <- divisibleBy can not be 0" % schema)

        if value % divisibleBy != 0:
            self._error("Value %(value)r field '%(fieldname)s' is not "
                        "divisible by '%(divisibleBy)s'.",
                        x.get(fieldname), fieldname, divisibleBy=divisibleBy)

    def validate_disallow(self, x, fieldname, schema, disallow=None):
        '''
        Validates that the value of the given field does not match the
        disallowed type.
        '''
        try:
            self.validate_type(x, fieldname, schema, disallow)
        except ValidationError:
            return
        self._error("Value %(value)r of type %(disallow)s is disallowed for "
                    "field '%(fieldname)s'",
                    x.get(fieldname), fieldname, disallow=disallow)

    def validate(self, data, schema):
        '''
        Validates a piece of json data against the provided json-schema.
        '''
        self._validate(data, schema)

    def _validate(self, data, schema):
        self.__validate("_data", {"_data": data}, schema)

    def __validate(self, fieldname, data, schema):

        if schema is not None:
            if not isinstance(schema, dict):
                raise SchemaError("Schema structure is invalid.")

            newschema = copy.copy(schema)

            # handle 'optional', replace it with 'required'
            if 'required' in schema and 'optional' in schema:
                raise SchemaError('cannot specify optional and required')
            elif 'optional' in schema:
                warnings.warn('The "optional" attribute has been replaced by "required"', DeprecationWarning)
                newschema['required'] = not schema['optional']
            elif 'required' not in schema:
                newschema['required'] = self.required_by_default

            if 'blank' not in schema:
                newschema['blank'] = self.blank_by_default

            for schemaprop in newschema:

                validatorname = "validate_" + schemaprop

                validator = getattr(self, validatorname, None)
                if validator:
                    validator(data, fieldname, schema,
                              newschema.get(schemaprop))

        return data

__all__ = ['SchemaValidator']