<s> import re <EOL> import sys <EOL> import copy <EOL> import socket <EOL> from datetime import datetime <EOL> from decimal import Decimal <EOL> from collections import Mapping , Container <EOL> if sys . version_info [ <NUM_LIT:0> ] == <NUM_LIT:3> : <EOL> _str_type = str <EOL> _int_types = ( int , ) <EOL> else : <EOL> _str_type = basestring <EOL> _int_types = ( int , long ) <EOL> class SchemaError ( ValueError ) : <EOL> """<STR_LIT>""" <EOL> class ValidationError ( ValueError ) : <EOL> """<STR_LIT>""" <EOL> class FieldValidationError ( ValidationError ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , message , fieldname , value , path = '<STR_LIT>' ) : <EOL> message = "<STR_LIT>" . format ( value , path , message ) <EOL> super ( FieldValidationError , self ) . __init__ ( message ) <EOL> self . fieldname = fieldname <EOL> self . value = value <EOL> self . path = path <EOL> class DependencyValidationError ( ValidationError ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , message ) : <EOL> super ( DependencyValidationError , self ) . __init__ ( message ) <EOL> class RequiredFieldValidationError ( ValidationError ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , message ) : <EOL> super ( RequiredFieldValidationError , self ) . __init__ ( message ) <EOL> class MultipleValidationError ( ValidationError ) : <EOL> def __init__ ( self , errors ) : <EOL> msg = "<STR_LIT>" . format ( len ( errors ) , '<STR_LIT:\n>' . join ( str ( e ) for e in errors ) ) <EOL> super ( MultipleValidationError , self ) . __init__ ( msg ) <EOL> self . errors = errors <EOL> def _generate_datetime_validator ( format_option , dateformat_string ) : <EOL> def validate_format_datetime ( validator , fieldname , value , format_option ) : <EOL> try : <EOL> datetime . strptime ( value , dateformat_string ) <EOL> except : <EOL> msg = "<STR_LIT>" <EOL> raise FieldValidationError ( msg . format ( format_option = format_option ) , fieldname , value ) <EOL> return validate_format_datetime <EOL> validate_format_date_time = _generate_datetime_validator ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> validate_format_date = _generate_datetime_validator ( '<STR_LIT:date>' , '<STR_LIT>' ) <EOL> validate_format_time = _generate_datetime_validator ( '<STR_LIT:time>' , '<STR_LIT>' ) <EOL> def validate_format_utc_millisec ( validator , fieldname , value , format_option ) : <EOL> if not isinstance ( value , _int_types + ( float , Decimal ) ) or value <= <NUM_LIT:0> : <EOL> msg = "<STR_LIT>" <EOL> raise FieldValidationError ( msg , fieldname , value ) <EOL> def validate_format_ip_address ( validator , fieldname , value , format_option ) : <EOL> try : <EOL> socket . inet_aton ( value ) <EOL> ip = len ( value . split ( '<STR_LIT:.>' ) ) == <NUM_LIT:4> <EOL> except : <EOL> ip = False <EOL> if not ip : <EOL> msg = "<STR_LIT>" <EOL> raise FieldValidationError ( msg , fieldname , value ) <EOL> DEFAULT_FORMAT_VALIDATORS = { <EOL> '<STR_LIT>' : validate_format_date_time , <EOL> '<STR_LIT:date>' : validate_format_date , <EOL> '<STR_LIT:time>' : validate_format_time , <EOL> '<STR_LIT>' : validate_format_utc_millisec , <EOL> '<STR_LIT>' : validate_format_ip_address , <EOL> } <EOL> class SchemaValidator ( object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , format_validators = None , required_by_default = True , <EOL> blank_by_default = False , disallow_unknown_properties = False , <EOL> apply_default_to_data = False , fail_fast = True , <EOL> remove_unknown_properties = False ) : <EOL> self . _format_validators = { } <EOL> self . _errors = [ ] <EOL> for key , value in DEFAULT_FORMAT_VALIDATORS . items ( ) : <EOL> self . register_format_validator ( key , value ) <EOL> if format_validators : <EOL> for key , value in format_validators . items ( ) : <EOL> self . register_format_validator ( key , value ) <EOL> self . required_by_default = required_by_default <EOL> self . blank_by_default = blank_by_default <EOL> self . disallow_unknown_properties = disallow_unknown_properties <EOL> self . apply_default_to_data = apply_default_to_data <EOL> self . fail_fast = fail_fast <EOL> self . remove_unknown_properties = remove_unknown_properties <EOL> def register_format_validator ( self , format_name , format_validator_fun ) : <EOL> self . _format_validators [ format_name ] = format_validator_fun <EOL> def validate_type_string ( self , val ) : <EOL> return isinstance ( val , _str_type ) <EOL> def validate_type_integer ( self , val ) : <EOL> return type ( val ) in _int_types <EOL> def validate_type_number ( self , val ) : <EOL> return type ( val ) in _int_types + ( float , Decimal , ) <EOL> def validate_type_boolean ( self , val ) : <EOL> return type ( val ) == bool <EOL> def validate_type_object ( self , val ) : <EOL> return isinstance ( val , Mapping ) or ( hasattr ( val , '<STR_LIT>' ) and hasattr ( val , '<STR_LIT>' ) ) <EOL> def validate_type_array ( self , val ) : <EOL> return isinstance ( val , ( list , tuple ) ) <EOL> def validate_type_null ( self , val ) : <EOL> return val is None <EOL> def validate_type_any ( self , val ) : <EOL> return True <EOL> def _error ( self , desc , value , fieldname , exctype = FieldValidationError , path = '<STR_LIT>' , ** params ) : <EOL> params [ '<STR_LIT:value>' ] = value <EOL> params [ '<STR_LIT>' ] = fieldname <EOL> message = desc . format ( ** params ) <EOL> if exctype == FieldValidationError : <EOL> err = FieldValidationError ( message , fieldname , value , path ) <EOL> elif exctype == DependencyValidationError or exctype == RequiredFieldValidationError : <EOL> err = exctype ( message ) <EOL> err . fieldname = fieldname <EOL> err . path = path <EOL> if self . fail_fast : <EOL> raise err <EOL> else : <EOL> self . _errors . append ( err ) <EOL> def _validate_unknown_properties ( self , schema , data , fieldname ) : <EOL> """<STR_LIT>""" <EOL> schema_properties = set ( schema ) <EOL> data_properties = set ( data ) <EOL> delta = data_properties - schema_properties <EOL> if delta : <EOL> unknowns = '<STR_LIT:U+002CU+0020>' . join ( [ '<STR_LIT>' . format ( x ) for x in delta ] ) <EOL> raise SchemaError ( '<STR_LIT>' . format ( <EOL> fieldname = fieldname , unknowns = unknowns ) ) <EOL> def _remove_unknown_properties ( self , schema , data , fieldname ) : <EOL> """<STR_LIT>""" <EOL> schema_properties = set ( schema ) <EOL> data_properties = set ( data ) <EOL> unknown_fields = data_properties - schema_properties <EOL> for unknown_field in unknown_fields : <EOL> del data [ unknown_field ] <EOL> def validate_type ( self , x , fieldname , schema , path , fieldtype = None ) : <EOL> '''<STR_LIT>''' <EOL> fieldexists = True <EOL> try : <EOL> value = x [ fieldname ] <EOL> except KeyError : <EOL> fieldexists = False <EOL> value = None <EOL> if fieldtype and fieldexists : <EOL> if isinstance ( fieldtype , ( list , tuple ) ) : <EOL> datavalid = False <EOL> errorlist = [ ] <EOL> for eachtype in fieldtype : <EOL> try : <EOL> errors = self . _errors [ : ] <EOL> self . validate_type ( x , fieldname , eachtype , path , eachtype ) <EOL> if len ( self . _errors ) > len ( errors ) : <EOL> raise self . _errors . pop ( ) <EOL> datavalid = True <EOL> break <EOL> except ValidationError as err : <EOL> errorlist . append ( err ) <EOL> if not datavalid : <EOL> self . _error ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> value , fieldname , path = path , numsubtypes = len ( fieldtype ) , <EOL> fieldtype = fieldtype , errorlist = errorlist ) <EOL> elif isinstance ( fieldtype , dict ) : <EOL> try : <EOL> self . __validate ( fieldname , x , fieldtype , path ) <EOL> except ValueError as e : <EOL> raise e <EOL> else : <EOL> try : <EOL> type_checker = getattr ( self , '<STR_LIT>' + fieldtype ) <EOL> except AttributeError : <EOL> raise SchemaError ( "<STR_LIT>" . format ( fieldtype ) ) <EOL> if not type_checker ( value ) : <EOL> self . _error ( "<STR_LIT>" , value , fieldname , path = path , <EOL> fieldtype = fieldtype ) <EOL> def validate_properties ( self , x , fieldname , schema , path , properties = None ) : <EOL> '''<STR_LIT>''' <EOL> value = x . get ( fieldname ) <EOL> if value is not None : <EOL> if isinstance ( value , dict ) : <EOL> if isinstance ( properties , dict ) : <EOL> if self . disallow_unknown_properties : <EOL> self . _validate_unknown_properties ( properties , value , fieldname ) <EOL> elif self . remove_unknown_properties : <EOL> self . _remove_unknown_properties ( properties , value , fieldname ) <EOL> for property in properties : <EOL> self . __validate ( property , value , properties . get ( property ) , <EOL> path + '<STR_LIT:.>' + property ) <EOL> else : <EOL> raise SchemaError ( "<STR_LIT>" <EOL> . format ( fieldname ) ) <EOL> def validate_items ( self , x , fieldname , schema , path , items = None ) : <EOL> '''<STR_LIT>''' <EOL> if x . get ( fieldname ) is not None : <EOL> value = x . get ( fieldname ) <EOL> if isinstance ( value , ( list , tuple ) ) : <EOL> if isinstance ( items , ( list , tuple ) ) : <EOL> if '<STR_LIT>' not in schema and len ( items ) != len ( value ) : <EOL> self . _error ( "<STR_LIT>" , value , fieldname , <EOL> path = path ) <EOL> else : <EOL> for index , item in enumerate ( items ) : <EOL> try : <EOL> self . __validate ( "<STR_LIT>" , { "<STR_LIT>" : value [ index ] } , item , <EOL> '<STR_LIT>' . format ( path , index ) ) <EOL> except FieldValidationError as e : <EOL> raise type ( e ) ( "<STR_LIT>" % <EOL> ( fieldname , e ) , fieldname , e . value ) <EOL> elif isinstance ( items , dict ) : <EOL> for index , item in enumerate ( value ) : <EOL> if self . disallow_unknown_properties and '<STR_LIT>' in items : <EOL> self . _validate_unknown_properties ( items [ '<STR_LIT>' ] , item , <EOL> fieldname ) <EOL> elif self . remove_unknown_properties and '<STR_LIT>' in items : <EOL> self . _remove_unknown_properties ( items [ '<STR_LIT>' ] , item , <EOL> fieldname ) <EOL> self . __validate ( "<STR_LIT>" , { "<STR_LIT>" : item } , items , <EOL> '<STR_LIT>' . format ( path , index ) ) <EOL> else : <EOL> raise SchemaError ( "<STR_LIT>" <EOL> "<STR_LIT>" . format ( fieldname ) ) <EOL> def validate_required ( self , x , fieldname , schema , path , required ) : <EOL> '''<STR_LIT>''' <EOL> if fieldname not in x and required : <EOL> self . _error ( "<STR_LIT>" , None , fieldname , path = path , <EOL> exctype = RequiredFieldValidationError ) <EOL> def validate_blank ( self , x , fieldname , schema , path , blank = False ) : <EOL> '''<STR_LIT>''' <EOL> value = x . get ( fieldname ) <EOL> if isinstance ( value , _str_type ) and not blank and not value : <EOL> self . _error ( "<STR_LIT>" , value , fieldname , path = path ) <EOL> def validate_patternProperties ( self , x , fieldname , schema , path , patternproperties = None ) : <EOL> if patternproperties is None : <EOL> patternproperties = { } <EOL> value_obj = x . get ( fieldname , { } ) <EOL> for pattern , schema in patternproperties . items ( ) : <EOL> for key , value in value_obj . items ( ) : <EOL> if re . match ( pattern , key ) : <EOL> self . __validate ( "<STR_LIT>" , { "<STR_LIT>" : value } , schema , path ) <EOL> def validate_additionalItems ( self , x , fieldname , schema , path , additionalItems = False ) : <EOL> value = x . get ( fieldname ) <EOL> if not isinstance ( value , ( list , tuple ) ) : <EOL> return <EOL> if isinstance ( additionalItems , bool ) : <EOL> if additionalItems or '<STR_LIT>' not in schema : <EOL> return <EOL> elif len ( value ) != len ( schema [ '<STR_LIT>' ] ) : <EOL> self . _error ( "<STR_LIT>" , value , fieldname , path = path ) <EOL> remaining = value [ len ( schema [ '<STR_LIT>' ] ) : ] <EOL> if len ( remaining ) > <NUM_LIT:0> : <EOL> self . __validate ( "<STR_LIT>" , { "<STR_LIT>" : remaining } , { "<STR_LIT>" : additionalItems } , path ) <EOL> def validate_additionalProperties ( self , x , fieldname , schema , path , additionalProperties = None ) : <EOL> '''<STR_LIT>''' <EOL> value = x . get ( fieldname ) <EOL> if not isinstance ( value , dict ) : <EOL> return <EOL> if isinstance ( additionalProperties , bool ) and additionalProperties : <EOL> return <EOL> value = x . get ( fieldname ) <EOL> if isinstance ( additionalProperties , ( dict , bool ) ) : <EOL> properties = schema . get ( "<STR_LIT>" ) <EOL> patterns = schema [ "<STR_LIT>" ] . keys ( ) if '<STR_LIT>' in schema else [ ] <EOL> if properties is None : <EOL> properties = { } <EOL> if value is None : <EOL> value = { } <EOL> for eachProperty in value : <EOL> if ( eachProperty not in properties and not <EOL> any ( re . match ( p , eachProperty ) for p in patterns ) ) : <EOL> if additionalProperties is False : <EOL> self . _error ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , value , fieldname , prop = eachProperty , path = path ) <EOL> self . __validate ( eachProperty , value , additionalProperties , path ) <EOL> else : <EOL> raise SchemaError ( "<STR_LIT>" <EOL> "<STR_LIT>" . format ( fieldname ) ) <EOL> def validate_dependencies ( self , x , fieldname , schema , path , dependencies = None ) : <EOL> if x . get ( fieldname ) is not None : <EOL> if isinstance ( dependencies , _str_type ) : <EOL> dependencies = [ dependencies ] <EOL> if isinstance ( dependencies , ( list , tuple ) ) : <EOL> for dependency in dependencies : <EOL> if dependency not in x : <EOL> self . _error ( "<STR_LIT>" , <EOL> None , fieldname , dependency = dependency , path = path , <EOL> exctype = DependencyValidationError ) <EOL> elif isinstance ( dependencies , dict ) : <EOL> for k , v in dependencies . items ( ) : <EOL> if k in x and v not in x : <EOL> self . _error ( "<STR_LIT>" , None , fieldname , <EOL> k = k , v = v , exctype = DependencyValidationError , path = path ) <EOL> else : <EOL> raise SchemaError ( "<STR_LIT>" ) <EOL> def validate_minimum ( self , x , fieldname , schema , path , minimum = None ) : <EOL> '''<STR_LIT>''' <EOL> exclusive = schema . get ( '<STR_LIT>' , False ) <EOL> if x . get ( fieldname ) is not None : <EOL> value = x . get ( fieldname ) <EOL> if value is not None : <EOL> if ( type ( value ) in ( int , float ) and <EOL> ( not exclusive and value < minimum ) or <EOL> ( exclusive and value <= minimum ) ) : <EOL> self . _error ( "<STR_LIT>" , value , fieldname , <EOL> minimum = minimum , path = path ) <EOL> def validate_maximum ( self , x , fieldname , schema , path , maximum = None ) : <EOL> '''<STR_LIT>''' <EOL> exclusive = schema . get ( '<STR_LIT>' , False ) <EOL> if x . get ( fieldname ) is not None : <EOL> value = x . get ( fieldname ) <EOL> if value is not None : <EOL> if ( type ( value ) in ( int , float ) and <EOL> ( not exclusive and value > maximum ) or <EOL> ( exclusive and value >= maximum ) ) : <EOL> self . _error ( "<STR_LIT>" , value , fieldname , <EOL> maximum = maximum , path = path ) <EOL> def validate_maxLength ( self , x , fieldname , schema , path , length = None ) : <EOL> '''<STR_LIT>''' <EOL> value = x . get ( fieldname ) <EOL> if isinstance ( value , ( _str_type , list , tuple ) ) and len ( value ) > length : <EOL> self . _error ( "<STR_LIT>" , value , fieldname , <EOL> length = length , path = path ) <EOL> def validate_minLength ( self , x , fieldname , schema , path , length = None ) : <EOL> '''<STR_LIT>''' <EOL> value = x . get ( fieldname ) <EOL> if isinstance ( value , ( _str_type , list , tuple ) ) and len ( value ) < length : <EOL> self . _error ( "<STR_LIT>" , value , fieldname , <EOL> length = length , path = path ) <EOL> validate_minItems = validate_minLength <EOL> validate_maxItems = validate_maxLength <EOL> def validate_format ( self , x , fieldname , schema , path , format_option = None ) : <EOL> '''<STR_LIT>''' <EOL> value = x . get ( fieldname , None ) <EOL> format_validator = self . _format_validators . get ( format_option , None ) <EOL> if format_validator and value is not None : <EOL> try : <EOL> format_validator ( self , fieldname , value , format_option ) <EOL> except FieldValidationError as fve : <EOL> if self . fail_fast : <EOL> raise <EOL> else : <EOL> self . _errors . append ( fve ) <EOL> def validate_pattern ( self , x , fieldname , schema , path , pattern = None ) : <EOL> '''<STR_LIT>''' <EOL> value = x . get ( fieldname ) <EOL> if ( isinstance ( value , _str_type ) and ( isinstance ( pattern , _str_type ) and not <EOL> re . match ( pattern , value ) or not isinstance ( pattern , _str_type ) and not <EOL> pattern . match ( value ) ) ) : <EOL> self . _error ( "<STR_LIT>" , value , fieldname , <EOL> pattern = pattern , path = path ) <EOL> def validate_uniqueItems ( self , x , fieldname , schema , path , uniqueItems = False ) : <EOL> '''<STR_LIT>''' <EOL> if isinstance ( uniqueItems , bool ) and not uniqueItems : <EOL> return <EOL> values = x . get ( fieldname ) <EOL> if not isinstance ( values , ( list , tuple ) ) : <EOL> return <EOL> hashables = set ( ) <EOL> unhashables = [ ] <EOL> for value in values : <EOL> if isinstance ( value , ( list , dict ) ) : <EOL> container , add = unhashables , unhashables . append <EOL> else : <EOL> container , add = hashables , hashables . add <EOL> if value in container : <EOL> self . _error ( "<STR_LIT>" , value , fieldname , path = path ) <EOL> else : <EOL> add ( value ) <EOL> def validate_enum ( self , x , fieldname , schema , path , options = None ) : <EOL> '''<STR_LIT>''' <EOL> value = x . get ( fieldname ) <EOL> if value is not None : <EOL> if callable ( options ) : <EOL> options = options ( x ) <EOL> if not isinstance ( options , Container ) : <EOL> raise SchemaError ( "<STR_LIT>" . format ( <EOL> options , fieldname ) ) <EOL> if value not in options : <EOL> if not ( value == '<STR_LIT>' and schema . get ( '<STR_LIT:blank>' , self . blank_by_default ) ) : <EOL> self . _error ( "<STR_LIT>" , value , fieldname , <EOL> options = options , path = path ) <EOL> def validate_title ( self , x , fieldname , schema , path , title = None ) : <EOL> if not isinstance ( title , ( _str_type , type ( None ) ) ) : <EOL> raise SchemaError ( "<STR_LIT>" . format ( fieldname ) ) <EOL> def validate_description ( self , x , fieldname , schema , path , description = None ) : <EOL> if not isinstance ( description , ( _str_type , type ( None ) ) ) : <EOL> raise SchemaError ( "<STR_LIT>" . format ( fieldname ) ) <EOL> def validate_divisibleBy ( self , x , fieldname , schema , path , divisibleBy = None ) : <EOL> value = x . get ( fieldname ) <EOL> if not self . validate_type_number ( value ) : <EOL> return <EOL> if divisibleBy == <NUM_LIT:0> : <EOL> raise SchemaError ( "<STR_LIT>" . format ( schema ) ) <EOL> if value % divisibleBy != <NUM_LIT:0> : <EOL> self . _error ( "<STR_LIT>" , x . get ( fieldname ) , fieldname , <EOL> divisibleBy = divisibleBy , path = path ) <EOL> def validate_disallow ( self , x , fieldname , schema , path , disallow = None ) : <EOL> '''<STR_LIT>''' <EOL> try : <EOL> self . validate_type ( x , fieldname , schema , path , disallow ) <EOL> except ValidationError : <EOL> return <EOL> self . _error ( "<STR_LIT>" , x . get ( fieldname ) , fieldname , <EOL> disallow = disallow , path = path ) <EOL> def validate ( self , data , schema ) : <EOL> '''<STR_LIT>''' <EOL> self . __validate ( "<STR_LIT:data>" , { "<STR_LIT:data>" : data } , schema , '<STR_LIT>' ) <EOL> if self . _errors : <EOL> raise MultipleValidationError ( self . _errors ) <EOL> def __validate ( self , fieldname , data , schema , path ) : <EOL> if schema is not None : <EOL> if not isinstance ( schema , dict ) : <EOL> raise SchemaError ( "<STR_LIT>" % <EOL> ( fieldname , type ( schema ) . __name__ ) ) <EOL> add_required_rule = self . required_by_default and '<STR_LIT>' not in schema <EOL> add_not_blank_rule = not self . blank_by_default and '<STR_LIT:blank>' not in schema <EOL> if add_required_rule or add_not_blank_rule : <EOL> newschema = copy . copy ( schema ) <EOL> if add_required_rule : <EOL> newschema [ '<STR_LIT>' ] = self . required_by_default <EOL> if add_not_blank_rule : <EOL> newschema [ '<STR_LIT:blank>' ] = self . blank_by_default <EOL> else : <EOL> newschema = schema <EOL> if self . apply_default_to_data and '<STR_LIT:default>' in schema : <EOL> try : <EOL> self . validate_type ( x = { '<STR_LIT>' : schema [ '<STR_LIT:default>' ] } , fieldname = '<STR_LIT>' , <EOL> schema = schema , <EOL> fieldtype = schema [ '<STR_LIT:type>' ] if '<STR_LIT:type>' in schema else None , <EOL> path = path ) <EOL> except FieldValidationError as exc : <EOL> raise SchemaError ( exc ) <EOL> if fieldname not in data : <EOL> data [ fieldname ] = schema [ '<STR_LIT:default>' ] <EOL> for schemaprop in newschema : <EOL> validatorname = "<STR_LIT>" + schemaprop <EOL> validator = getattr ( self , validatorname , None ) <EOL> if validator : <EOL> validator ( data , fieldname , schema , path , newschema . get ( schemaprop ) ) <EOL> return data </s>
