<s> """<STR_LIT>""" <EOL> import warnings <EOL> from pyschema import core <EOL> from pyschema . types import Field , Boolean , Integer , Float <EOL> from pyschema . types import Bytes , Text , Enum , List , Map , SubRecord <EOL> try : <EOL> import simplejson as json <EOL> except ImportError : <EOL> import json <EOL> Boolean . avro_type_name = "<STR_LIT>" <EOL> Bytes . avro_type_name = "<STR_LIT>" <EOL> Text . avro_type_name = "<STR_LIT:string>" <EOL> Enum . avro_type_name = "<STR_LIT>" <EOL> List . avro_type_name = "<STR_LIT>" <EOL> Map . avro_type_name = "<STR_LIT>" <EOL> @ Float . mixin <EOL> class FloatMixin : <EOL> @ property <EOL> def avro_type_name ( self ) : <EOL> if self . size <= <NUM_LIT:4> : <EOL> return '<STR_LIT:float>' <EOL> return '<STR_LIT>' <EOL> @ Integer . mixin <EOL> class IntegerMixin : <EOL> @ property <EOL> def avro_type_name ( self ) : <EOL> if self . size <= <NUM_LIT:4> : <EOL> return '<STR_LIT:int>' <EOL> return '<STR_LIT>' <EOL> @ Field . mixin <EOL> class FieldMixin : <EOL> def avro_type_schema ( self , state ) : <EOL> """<STR_LIT>""" <EOL> simple_type = self . simplified_avro_type_schema ( state ) <EOL> if self . nullable : <EOL> if self . default in ( None , core . NO_DEFAULT ) : <EOL> return [ "<STR_LIT:null>" , simple_type ] <EOL> else : <EOL> return [ simple_type , "<STR_LIT:null>" ] <EOL> else : <EOL> return simple_type <EOL> def simplified_avro_type_schema ( self , state ) : <EOL> """<STR_LIT>""" <EOL> return self . avro_type_name <EOL> def avro_dump ( self , o ) : <EOL> if o is None : <EOL> return None <EOL> else : <EOL> if self . nullable : <EOL> return { self . avro_type_name : self . dump ( o ) } <EOL> else : <EOL> return self . dump ( o ) <EOL> def avro_load ( self , o ) : <EOL> if o is None : <EOL> return None <EOL> else : <EOL> if self . nullable : <EOL> return self . load ( o [ self . avro_type_name ] ) <EOL> else : <EOL> return self . load ( o ) <EOL> def avro_default_value ( self ) : <EOL> return self . default <EOL> @ List . mixin <EOL> class ListMixin : <EOL> def simplified_avro_type_schema ( self , state ) : <EOL> return { <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : self . field_type . avro_type_schema ( state ) <EOL> } <EOL> def avro_dump ( self , obj ) : <EOL> if obj is None : <EOL> return None <EOL> else : <EOL> l = [ self . field_type . avro_dump ( o ) for o in obj ] <EOL> if self . nullable : <EOL> return { self . avro_type_name : l } <EOL> else : <EOL> return l <EOL> def avro_load ( self , obj ) : <EOL> if obj is None : <EOL> return None <EOL> else : <EOL> if self . nullable : <EOL> obj = obj [ self . avro_type_name ] <EOL> return [ <EOL> self . field_type . avro_load ( o ) <EOL> for o in obj <EOL> ] <EOL> @ Enum . mixin <EOL> class EnumMixin : <EOL> @ property <EOL> def avro_type_name ( self ) : <EOL> if self . name is None : <EOL> return "<STR_LIT>" <EOL> return self . name <EOL> def simplified_avro_type_schema ( self , state ) : <EOL> return { <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT:name>" : self . avro_type_name , <EOL> "<STR_LIT>" : list ( self . values ) <EOL> } <EOL> @ SubRecord . mixin <EOL> class SubRecordMixin : <EOL> def simplified_avro_type_schema ( self , state ) : <EOL> return get_schema_dict ( self . _schema , state ) <EOL> @ property <EOL> def avro_type_name ( self ) : <EOL> return core . get_full_name ( self . _schema ) <EOL> def avro_dump ( self , obj ) : <EOL> if obj is None : <EOL> return None <EOL> if self . nullable : <EOL> return { self . avro_type_name : to_json_compatible ( obj ) } <EOL> else : <EOL> return to_json_compatible ( obj ) <EOL> def _get_record_data ( self , obj ) : <EOL> """<STR_LIT>""" <EOL> if self . avro_type_name in obj : <EOL> return obj [ self . avro_type_name ] <EOL> else : <EOL> base_name = self . avro_type_name . split ( '<STR_LIT:.>' ) [ - <NUM_LIT:1> ] <EOL> return obj [ base_name ] <EOL> def avro_load ( self , obj ) : <EOL> if obj is None : <EOL> return None <EOL> if self . nullable : <EOL> return from_json_compatible ( <EOL> self . _schema , <EOL> self . _get_record_data ( obj ) <EOL> ) <EOL> else : <EOL> return from_json_compatible ( <EOL> self . _schema , <EOL> obj <EOL> ) <EOL> @ Map . mixin <EOL> class MapMixin : <EOL> def simplified_avro_type_schema ( self , state ) : <EOL> assert isinstance ( self . key_type , Text ) <EOL> return { <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : self . value_type . avro_type_schema ( state ) <EOL> } <EOL> def avro_dump ( self , obj ) : <EOL> if obj is None : <EOL> return None <EOL> else : <EOL> m = dict ( [ ( <EOL> self . key_type . dump ( k ) , <EOL> self . value_type . avro_dump ( v ) <EOL> ) for k , v in obj . iteritems ( ) ] ) <EOL> if self . nullable : <EOL> return { self . avro_type_name : m } <EOL> else : <EOL> return m <EOL> def avro_load ( self , obj ) : <EOL> if obj is None : <EOL> return None <EOL> else : <EOL> if self . nullable : <EOL> obj = obj [ self . avro_type_name ] <EOL> m = dict ( [ ( <EOL> self . key_type . load ( k ) , <EOL> self . value_type . avro_load ( v ) <EOL> ) for k , v in obj . iteritems ( ) ] ) <EOL> return m <EOL> class SchemaGeneratorState ( object ) : <EOL> def __init__ ( self ) : <EOL> self . declared_records = set ( ) <EOL> def get_schema_dict ( record , state = None ) : <EOL> state = state or SchemaGeneratorState ( ) <EOL> full_name = core . get_full_name ( record ) <EOL> if full_name in state . declared_records : <EOL> return full_name <EOL> state . declared_records . add ( full_name ) <EOL> if '<STR_LIT:.>' in full_name : <EOL> namespace , _ = full_name . rsplit ( '<STR_LIT:.>' , <NUM_LIT:1> ) <EOL> else : <EOL> namespace = None <EOL> avro_record = { <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT:name>" : record . _schema_name , <EOL> } <EOL> if namespace : <EOL> avro_record [ "<STR_LIT>" ] = namespace <EOL> if record . __doc__ is not None : <EOL> avro_record [ "<STR_LIT>" ] = record . __doc__ <EOL> avro_fields = [ ] <EOL> for field_name , field_type in record . _fields . iteritems ( ) : <EOL> field_spec = { <EOL> "<STR_LIT:name>" : field_name , <EOL> "<STR_LIT:type>" : field_type . avro_type_schema ( state ) <EOL> } <EOL> if field_type . default is not core . NO_DEFAULT : <EOL> field_spec [ "<STR_LIT:default>" ] = field_type . avro_default_value ( ) <EOL> if field_type . description is not None : <EOL> field_spec [ "<STR_LIT>" ] = field_type . description <EOL> avro_fields . append ( field_spec ) <EOL> avro_record [ "<STR_LIT>" ] = avro_fields <EOL> return avro_record <EOL> def get_schema_string ( record ) : <EOL> return json . dumps ( get_schema_dict ( record ) ) <EOL> def dumps ( record ) : <EOL> return json . dumps ( to_json_compatible ( record ) ) <EOL> def to_json_compatible ( record ) : <EOL> dct = { } <EOL> for name , fieldtype in record . _fields . iteritems ( ) : <EOL> value = getattr ( record , name ) <EOL> dct [ name ] = fieldtype . avro_dump ( value ) <EOL> return dct <EOL> def from_json_compatible ( schema , dct ) : <EOL> "<STR_LIT>" <EOL> kwargs = { } <EOL> for key in dct : <EOL> field_type = schema . _fields . get ( key ) <EOL> if field_type is None : <EOL> warnings . warn ( "<STR_LIT>" % ( schema . __name__ , key ) ) <EOL> continue <EOL> kwargs [ key ] = field_type . avro_load ( dct [ key ] ) <EOL> return schema ( ** kwargs ) <EOL> def loads ( <EOL> s , <EOL> record_store = None , <EOL> schema = None , <EOL> record_class = None <EOL> ) : <EOL> if record_class is not None : <EOL> warnings . warn ( <EOL> "<STR_LIT>" , <EOL> DeprecationWarning , <EOL> stacklevel = <NUM_LIT:2> <EOL> ) <EOL> schema = record_class <EOL> return core . loads ( s , record_store , schema , from_json_compatible ) </s>
