<s> """<STR_LIT>""" <EOL> import datetime <EOL> from decimal import Decimal <EOL> import copy <EOL> import json <EOL> import inspect as python_inspect <EOL> import iso8601 <EOL> from sqlalchemy import inspect <EOL> from sqlalchemy . orm . attributes import flag_modified <EOL> from sqlalchemy . orm . attributes import set_attribute <EOL> import jsonpointer <EOL> class BadJSONData ( Exception ) : <EOL> """<STR_LIT>""" <EOL> class BadStoredData ( Exception ) : <EOL> """<STR_LIT>""" <EOL> class CannotLookupData ( Exception ) : <EOL> """<STR_LIT>""" <EOL> class CannotProcessISO8601 ( Exception ) : <EOL> """<STR_LIT>""" <EOL> class NullConverter ( object ) : <EOL> def serialize ( self , val ) : <EOL> return val <EOL> def deserialize ( self , val ) : <EOL> return val <EOL> class ISO8601DatetimeConverter ( object ) : <EOL> """<STR_LIT>""" <EOL> def serialize ( self , val ) : <EOL> if val is None : <EOL> return val <EOL> assert isinstance ( val , datetime . datetime ) , "<STR_LIT>" . format ( type ( val ) ) <EOL> if not val . tzinfo : <EOL> raise CannotProcessISO8601 ( "<STR_LIT>" . format ( val ) ) <EOL> return val . isoformat ( ) <EOL> def deserialize ( self , val ) : <EOL> if val is None : <EOL> return val <EOL> return iso8601 . parse_date ( val ) <EOL> _marker = object ( ) <EOL> class JSONBProperty ( object ) : <EOL> """<STR_LIT>""" <EOL> UNDEFINED = jsonpointer . _nothing <EOL> def __init__ ( self , data_field , pointer , graceful = _marker , converter = NullConverter ) : <EOL> self . data_field = data_field <EOL> self . pointer = pointer <EOL> if type ( converter ) == type : <EOL> converter = converter ( ) <EOL> self . converter = converter <EOL> self . graceful = graceful <EOL> def ensure_valid_data ( self , obj ) : <EOL> """<STR_LIT>""" <EOL> if obj is None : <EOL> return None <EOL> field = getattr ( obj , self . data_field ) <EOL> if field is None : <EOL> insp = inspect ( obj ) <EOL> if insp . persistent or insp . detached : <EOL> raise BadStoredData ( "<STR_LIT>" . format ( self . data_field ) ) <EOL> else : <EOL> column = getattr ( obj . __class__ , self . data_field ) <EOL> default = column . default . arg <EOL> setattr ( obj , self . data_field , default ) <EOL> return default <EOL> return field <EOL> def is_graceful ( self ) : <EOL> return self . graceful != _marker <EOL> def __get__ ( self , obj , objtype = None ) : <EOL> data = self . ensure_valid_data ( obj ) <EOL> if data is None : <EOL> return None <EOL> try : <EOL> val = jsonpointer . resolve_pointer ( data , self . pointer , default = JSONBProperty . UNDEFINED ) <EOL> except jsonpointer . JsonPointerException as e : <EOL> if self . is_graceful ( ) : <EOL> val = self . graceful <EOL> else : <EOL> raise CannotLookupData ( "<STR_LIT>" . format ( self . pointer , obj ) ) <EOL> return self . converter . deserialize ( val ) <EOL> def __set__ ( self , obj , val ) : <EOL> data = copy . deepcopy ( self . ensure_valid_data ( obj ) ) <EOL> val = self . converter . serialize ( val ) <EOL> if val is not None : <EOL> if type ( val ) not in ( str , float , bool , int , dict ) : <EOL> raise BadJSONData ( "<STR_LIT>" . format ( self . pointer , type ( val ) ) ) <EOL> jsonpointer . set_pointer ( data , self . pointer , val ) <EOL> set_attribute ( obj , self . data_field , data ) <EOL> @ classmethod <EOL> def is_json_property ( cls , obj , name ) : <EOL> """<STR_LIT>""" <EOL> attr = python_inspect . getattr_static ( obj , name ) <EOL> return isinstance ( attr , JSONBProperty ) <EOL> class _DecimalEncoder ( json . JSONEncoder ) : <EOL> def default ( self , o ) : <EOL> if isinstance ( o , Decimal ) : <EOL> return str ( o ) <EOL> return super ( _DecimalEncoder , self ) . default ( o ) <EOL> def complex_json_dumps ( d ) : <EOL> """<STR_LIT>""" <EOL> return json . dumps ( d , cls = _DecimalEncoder ) </s>
