<s> """<STR_LIT>""" <EOL> import json <EOL> import logging <EOL> from openrcv import streams <EOL> from openrcv import utils <EOL> from openrcv . utils import PathInfo , ReprMixin , ENCODING_JSON <EOL> log = logging . getLogger ( __name__ ) <EOL> LIST_TYPES = ( list , tuple , type ( <NUM_LIT:0> for i in ( ) ) ) <EOL> class JsNull ( object ) : <EOL> pass <EOL> JS_NULL = JsNull ( ) <EOL> def call_json ( json_func , * args , ** kwargs ) : <EOL> return json_func ( * args , indent = <NUM_LIT:4> , sort_keys = True , ** kwargs ) <EOL> def to_json ( jsobj ) : <EOL> """<STR_LIT>""" <EOL> return call_json ( json . dumps , jsobj ) <EOL> def read_json_path ( path ) : <EOL> """<STR_LIT>""" <EOL> stream_info = JsonPathInfo ( path ) <EOL> with stream_info . open ( ) as f : <EOL> jsobj = json . load ( f ) <EOL> return jsobj <EOL> def write_json ( obj , resource = None , path = None ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> jsobj = obj . to_jsobj ( ) <EOL> except AttributeError : <EOL> jsobj = obj <EOL> if path is not None : <EOL> assert resource is None <EOL> resource = streams . FilePathResource ( path , encoding = ENCODING_JSON ) <EOL> with resource . open_write ( ) as f : <EOL> return call_json ( json . dump , jsobj , f ) <EOL> def from_model ( obj , cls ) : <EOL> """<STR_LIT>""" <EOL> if cls is None : <EOL> return obj <EOL> if isinstance ( obj , LIST_TYPES ) : <EOL> return [ from_model ( o , cls ) for o in obj ] <EOL> return cls . from_model ( obj ) <EOL> def from_jsobj ( jsobj , cls = None ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( jsobj , LIST_TYPES ) : <EOL> return [ from_jsobj ( o , cls = cls ) for o in jsobj ] <EOL> if cls is not None : <EOL> return cls . from_jsobj ( jsobj ) <EOL> if jsobj is None : <EOL> return JS_NULL <EOL> return jsobj <EOL> def to_jsobj ( obj ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( obj , LIST_TYPES ) : <EOL> return [ to_jsobj ( o ) for o in obj ] <EOL> if obj . __class__ . __module__ == "<STR_LIT>" : <EOL> return obj <EOL> return obj . to_jsobj ( ) <EOL> class JsonPathInfo ( PathInfo ) : <EOL> def __init__ ( self , path ) : <EOL> super ( ) . __init__ ( path , encoding = ENCODING_JSON ) <EOL> class JsonableError ( Exception ) : <EOL> pass <EOL> class JsonDeserializeError ( Exception ) : <EOL> pass <EOL> class Attribute ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , name , cls = None , keyword = None , model = True ) : <EOL> """<STR_LIT>""" <EOL> if keyword is None : <EOL> keyword = name <EOL> self . cls = cls <EOL> self . keyword = keyword <EOL> self . model = model <EOL> self . name = name <EOL> def model_value ( self , model_obj ) : <EOL> """<STR_LIT>""" <EOL> return from_model ( getattr ( model_obj , self . name ) , self . cls ) <EOL> class JsonableMixin ( ReprMixin ) : <EOL> """<STR_LIT>""" <EOL> meta_attrs = ( ) <EOL> @ classmethod <EOL> def attrs ( cls ) : <EOL> return list ( cls . meta_attrs ) + list ( cls . data_attrs ) <EOL> @ classmethod <EOL> def model_attrs ( cls ) : <EOL> return ( attr for attr in cls . attrs ( ) if attr . model ) <EOL> @ classmethod <EOL> def keywords_to_attrs ( cls ) : <EOL> """<STR_LIT>""" <EOL> return { attr . keyword : attr for attr in cls . attrs ( ) } <EOL> @ classmethod <EOL> def keywords_to_init_defaults ( cls ) : <EOL> """<STR_LIT>""" <EOL> mapping = cls . keywords_to_attrs ( ) <EOL> return dict . fromkeys ( mapping . keys ( ) ) <EOL> @ classmethod <EOL> def model_to_kwargs ( cls , model_obj ) : <EOL> """<STR_LIT>""" <EOL> return { attr . keyword : attr . model_value ( model_obj ) for attr in cls . model_attrs ( ) } <EOL> def __init__ ( self , ** kwargs ) : <EOL> defaults = self . keywords_to_init_defaults ( ) <EOL> values = defaults . copy ( ) <EOL> values . update ( kwargs ) <EOL> if len ( values ) != len ( defaults ) : <EOL> valid = sorted ( defaults . keys ( ) ) <EOL> invalid = sorted ( set ( values . keys ( ) ) - set ( valid ) ) <EOL> raise TypeError ( "<STR_LIT>" . <EOL> format ( "<STR_LIT:U+002CU+0020>" . join ( repr ( k ) for k in invalid ) , <EOL> "<STR_LIT:U+002CU+0020>" . join ( valid ) ) ) <EOL> attrs = self . keywords_to_attrs ( ) <EOL> for keyword , value in values . items ( ) : <EOL> attr = attrs [ keyword ] <EOL> setattr ( self , attr . name , value ) <EOL> def repr_info ( self ) : <EOL> """<STR_LIT>""" <EOL> attrs = self . attrs ( ) <EOL> return "<STR_LIT:U+0020>" . join ( "<STR_LIT>" . format ( attr . name , getattr ( self , attr . name ) ) <EOL> for attr in attrs ) <EOL> def return_false ( self , * args ) : <EOL> return False <EOL> def raise_not_equal ( self , other , value_desc , self_value , other_value ) : <EOL> raise AssertionError ( "<STR_LIT>" . <EOL> format ( self , other , value_desc , self_value , other_value ) ) <EOL> def _check_eq ( self , other , not_equal_func ) : <EOL> if type ( self ) != type ( other ) : <EOL> return not_equal_func ( other , "<STR_LIT>" , type ( self ) , type ( other ) ) <EOL> for attr in self . attrs ( ) : <EOL> name = attr . name <EOL> if ( getattr ( self , name ) != getattr ( other , name ) ) : <EOL> return not_equal_func ( other , "<STR_LIT>" . format ( name ) , getattr ( self , name ) , <EOL> getattr ( other , name ) ) <EOL> return True <EOL> def assert_equal ( self , other ) : <EOL> """<STR_LIT>""" <EOL> return self . _check_eq ( other , self . raise_not_equal ) <EOL> def __eq__ ( self , other ) : <EOL> return self . _check_eq ( other , self . return_false ) <EOL> def __ne__ ( self , other ) : <EOL> return not self . __eq__ ( other ) <EOL> def _attrs_from_jsdict ( self , attrs , jsdict ) : <EOL> """<STR_LIT>""" <EOL> for attr in attrs : <EOL> name , cls = attr . name , attr . cls <EOL> try : <EOL> jsobj = jsdict [ name ] <EOL> except KeyError : <EOL> obj = None <EOL> else : <EOL> obj = from_jsobj ( jsobj , cls = cls ) <EOL> setattr ( self , name , obj ) <EOL> def _attrs_to_jsdict ( self , attrs , jsdict ) : <EOL> """<STR_LIT>""" <EOL> for attr in attrs : <EOL> try : <EOL> name , cls = attr . name , attr . cls <EOL> except AttributeError : <EOL> raise JsonableError ( "<STR_LIT>" % attr ) <EOL> try : <EOL> value = getattr ( self , name ) <EOL> except TypeError : <EOL> raise JsonableError ( "<STR_LIT>" % name ) <EOL> if value is None : <EOL> continue <EOL> jsobj = to_jsobj ( value ) <EOL> jsdict [ name ] = jsobj <EOL> def save_from_jsobj ( self , jsobj ) : <EOL> """<STR_LIT>""" <EOL> keys = set ( ) <EOL> try : <EOL> meta_dict = jsobj [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> pass <EOL> else : <EOL> keys |= set ( meta_dict . keys ( ) ) <EOL> self . _attrs_from_jsdict ( self . meta_attrs , meta_dict ) <EOL> keys |= set ( jsobj . keys ( ) ) <EOL> keys -= set ( ( '<STR_LIT>' , ) ) <EOL> extra_keys = keys - set ( ( attr . name for attr in self . attrs ( ) ) ) <EOL> if extra_keys : <EOL> log . warning ( "<STR_LIT>" % ( list ( extra_keys ) , jsobj ) ) <EOL> self . _attrs_from_jsdict ( self . data_attrs , jsobj ) <EOL> def get_meta_dict ( self ) : <EOL> """<STR_LIT>""" <EOL> meta = { } <EOL> self . _attrs_to_jsdict ( self . meta_attrs , meta ) <EOL> return meta <EOL> @ classmethod <EOL> def from_model ( cls , model_obj ) : <EOL> """<STR_LIT>""" <EOL> kwargs = cls . model_to_kwargs ( model_obj ) <EOL> jsonable = cls ( ** kwargs ) <EOL> jsonable . save_from_model ( model_obj ) <EOL> return jsonable <EOL> def save_from_model ( self , model_obj ) : <EOL> pass <EOL> def to_model ( self ) : <EOL> raise utils . NoImplementation ( self ) <EOL> @ classmethod <EOL> def from_jsobj ( cls , jsobj ) : <EOL> """<STR_LIT>""" <EOL> jsonable = cls ( ) <EOL> jsonable . save_from_jsobj ( jsobj ) <EOL> return jsonable <EOL> def to_jsobj ( self ) : <EOL> """<STR_LIT>""" <EOL> jsobj = { } <EOL> meta = self . get_meta_dict ( ) <EOL> if meta : <EOL> jsobj [ '<STR_LIT>' ] = meta <EOL> self . _attrs_to_jsdict ( self . data_attrs , jsobj ) <EOL> return jsobj <EOL> def to_json ( self ) : <EOL> """<STR_LIT>""" <EOL> return to_json ( self . to_jsobj ( ) ) </s>
