<s> import datetime <EOL> import decimal <EOL> from simplegeneric import generic <EOL> from wsme . exc import ClientSideError <EOL> from wsme . protocol import CallContext , Protocol , expose <EOL> from wsme . utils import parse_isodate , parse_isodatetime , parse_isotime <EOL> from wsme . rest . args import from_params <EOL> from wsme . types import iscomplex , isusertype , list_attributes , Unset <EOL> import wsme . types <EOL> try : <EOL> import simplejson as json <EOL> except ImportError : <EOL> import json <EOL> from six import u <EOL> class APIDefinitionGenerator ( object ) : <EOL> tpl = """<STR_LIT>""" <EOL> descriptor_tpl = """<STR_LIT>""" <EOL> provider_tpl = """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> pass <EOL> def render ( self , rootns , webpath , namespaces , fullns ) : <EOL> descriptors = u ( '<STR_LIT>' ) <EOL> for ns in sorted ( namespaces ) : <EOL> descriptors += self . descriptor_tpl % { <EOL> '<STR_LIT>' : ns , <EOL> '<STR_LIT>' : rootns , <EOL> '<STR_LIT>' : fullns ( ns ) , <EOL> '<STR_LIT>' : '<STR_LIT:\n>' . join ( ( <EOL> '<STR_LIT:U+0020>' * <NUM_LIT:4> + line <EOL> for line <EOL> in json . dumps ( namespaces [ ns ] , indent = <NUM_LIT:4> ) . split ( '<STR_LIT:\n>' ) <EOL> ) ) <EOL> } <EOL> providers = u ( '<STR_LIT>' ) <EOL> for ns in sorted ( namespaces ) : <EOL> providers += self . provider_tpl % { <EOL> '<STR_LIT>' : fullns ( ns ) <EOL> } <EOL> r = self . tpl % { <EOL> '<STR_LIT>' : rootns , <EOL> '<STR_LIT>' : webpath , <EOL> '<STR_LIT>' : descriptors , <EOL> '<STR_LIT>' : providers , <EOL> } <EOL> return r <EOL> @ generic <EOL> def fromjson ( datatype , value ) : <EOL> if value is None : <EOL> return None <EOL> if iscomplex ( datatype ) : <EOL> newvalue = datatype ( ) <EOL> for attrdef in list_attributes ( datatype ) : <EOL> if attrdef . name in value : <EOL> setattr ( newvalue , attrdef . key , <EOL> fromjson ( attrdef . datatype , value [ attrdef . name ] ) ) <EOL> value = newvalue <EOL> elif isusertype ( datatype ) : <EOL> value = datatype . frombasetype ( fromjson ( datatype . basetype , value ) ) <EOL> return value <EOL> @ generic <EOL> def tojson ( datatype , value ) : <EOL> if value is None : <EOL> return value <EOL> if iscomplex ( datatype ) : <EOL> d = { } <EOL> for attrdef in list_attributes ( datatype ) : <EOL> attrvalue = getattr ( value , attrdef . key ) <EOL> if attrvalue is not Unset : <EOL> d [ attrdef . name ] = tojson ( attrdef . datatype , attrvalue ) <EOL> value = d <EOL> elif isusertype ( datatype ) : <EOL> value = tojson ( datatype . basetype , datatype . tobasetype ( value ) ) <EOL> return value <EOL> @ fromjson . when_type ( wsme . types . ArrayType ) <EOL> def array_fromjson ( datatype , value ) : <EOL> return [ fromjson ( datatype . item_type , item ) for item in value ] <EOL> @ tojson . when_type ( wsme . types . ArrayType ) <EOL> def array_tojson ( datatype , value ) : <EOL> if value is None : <EOL> return value <EOL> return [ tojson ( datatype . item_type , item ) for item in value ] <EOL> @ fromjson . when_type ( wsme . types . DictType ) <EOL> def dict_fromjson ( datatype , value ) : <EOL> if value is None : <EOL> return value <EOL> return dict ( ( <EOL> ( fromjson ( datatype . key_type , key ) , <EOL> fromjson ( datatype . value_type , value ) ) <EOL> for key , value in value . items ( ) <EOL> ) ) <EOL> @ tojson . when_type ( wsme . types . DictType ) <EOL> def dict_tojson ( datatype , value ) : <EOL> if value is None : <EOL> return value <EOL> return dict ( ( <EOL> ( tojson ( datatype . key_type , key ) , <EOL> tojson ( datatype . value_type , value ) ) <EOL> for key , value in value . items ( ) <EOL> ) ) <EOL> @ tojson . when_object ( wsme . types . bytes ) <EOL> def bytes_tojson ( datatype , value ) : <EOL> if value is None : <EOL> return value <EOL> return value . decode ( '<STR_LIT:ascii>' ) <EOL> @ fromjson . when_object ( wsme . types . bytes ) <EOL> def bytes_fromjson ( datatype , value ) : <EOL> if value is not None : <EOL> value = value . encode ( '<STR_LIT:ascii>' ) <EOL> return value <EOL> @ fromjson . when_object ( wsme . types . text ) <EOL> def text_fromjson ( datatype , value ) : <EOL> if isinstance ( value , wsme . types . bytes ) : <EOL> return value . decode ( '<STR_LIT:utf-8>' ) <EOL> return value <EOL> @ fromjson . when_object ( datetime . time ) <EOL> def time_fromjson ( datatype , value ) : <EOL> if value is None or value == '<STR_LIT>' : <EOL> return None <EOL> return parse_isotime ( value ) <EOL> @ tojson . when_object ( datetime . time ) <EOL> def time_tojson ( datatype , value ) : <EOL> if value is None : <EOL> return value <EOL> return value . isoformat ( ) <EOL> @ fromjson . when_object ( datetime . date ) <EOL> def date_fromjson ( datatype , value ) : <EOL> if value is None or value == '<STR_LIT>' : <EOL> return None <EOL> return parse_isodate ( value ) <EOL> @ tojson . when_object ( datetime . date ) <EOL> def date_tojson ( datatype , value ) : <EOL> if value is None : <EOL> return value <EOL> return value . isoformat ( ) <EOL> @ fromjson . when_object ( datetime . datetime ) <EOL> def datetime_fromjson ( datatype , value ) : <EOL> if value is None or value == '<STR_LIT>' : <EOL> return None <EOL> return parse_isodatetime ( value ) <EOL> @ tojson . when_object ( datetime . datetime ) <EOL> def datetime_tojson ( datatype , value ) : <EOL> if value is None : <EOL> return value <EOL> return value . isoformat ( ) <EOL> @ fromjson . when_object ( decimal . Decimal ) <EOL> def decimal_fromjson ( datatype , value ) : <EOL> if value is None : <EOL> return value <EOL> return decimal . Decimal ( value ) <EOL> @ tojson . when_object ( decimal . Decimal ) <EOL> def decimal_tojson ( datatype , value ) : <EOL> if value is None : <EOL> return value <EOL> return str ( value ) <EOL> class ExtCallContext ( CallContext ) : <EOL> def __init__ ( self , request , namespace , calldata ) : <EOL> super ( ExtCallContext , self ) . __init__ ( request ) <EOL> self . namespace = namespace <EOL> self . tid = calldata [ '<STR_LIT>' ] <EOL> self . action = calldata [ '<STR_LIT:action>' ] <EOL> self . method = calldata [ '<STR_LIT>' ] <EOL> self . params = calldata [ '<STR_LIT:data>' ] <EOL> class FormExtCallContext ( CallContext ) : <EOL> def __init__ ( self , request , namespace ) : <EOL> super ( FormExtCallContext , self ) . __init__ ( request ) <EOL> self . namespace = namespace <EOL> self . tid = request . params [ '<STR_LIT>' ] <EOL> self . action = request . params [ '<STR_LIT>' ] <EOL> self . method = request . params [ '<STR_LIT>' ] <EOL> self . params = [ ] <EOL> class ExtDirectProtocol ( Protocol ) : <EOL> """<STR_LIT>""" <EOL> name = '<STR_LIT>' <EOL> displayname = '<STR_LIT>' <EOL> content_types = [ '<STR_LIT:application/json>' , '<STR_LIT>' ] <EOL> def __init__ ( self , namespace = '<STR_LIT>' , params_notation = '<STR_LIT>' , nsfolder = None ) : <EOL> self . namespace = namespace <EOL> self . appns , self . apins = namespace . rsplit ( '<STR_LIT:.>' , <NUM_LIT:2> ) if '<STR_LIT:.>' in namespace else ( namespace , '<STR_LIT>' ) <EOL> self . default_params_notation = params_notation <EOL> self . appnsfolder = nsfolder <EOL> @ property <EOL> def api_alias ( self ) : <EOL> if self . appnsfolder : <EOL> alias = '<STR_LIT>' % ( <EOL> self . appnsfolder , <EOL> self . apins . replace ( '<STR_LIT:.>' , '<STR_LIT:/>' ) ) <EOL> return alias <EOL> def accept ( self , req ) : <EOL> path = req . path <EOL> assert path . startswith ( self . root . _webpath ) <EOL> path = path [ len ( self . root . _webpath ) : ] <EOL> return ( <EOL> path == self . api_alias or <EOL> path == "<STR_LIT>" or <EOL> path . startswith ( "<STR_LIT>" ) <EOL> ) <EOL> def iter_calls ( self , req ) : <EOL> path = req . path <EOL> assert path . startswith ( self . root . _webpath ) <EOL> path = path [ len ( self . root . _webpath ) : ] . strip ( ) <EOL> assert path . startswith ( '<STR_LIT>' ) , path <EOL> path = path [ <NUM_LIT> : ] . strip ( '<STR_LIT:/>' ) <EOL> if path : <EOL> namespace = path . split ( '<STR_LIT:.>' ) <EOL> else : <EOL> namespace = [ ] <EOL> if '<STR_LIT>' in req . params : <EOL> req . wsme_extdirect_batchcall = False <EOL> yield FormExtCallContext ( req , namespace ) <EOL> else : <EOL> data = json . loads ( req . body . decode ( '<STR_LIT:utf8>' ) ) <EOL> req . wsme_extdirect_batchcall = isinstance ( data , list ) <EOL> if not req . wsme_extdirect_batchcall : <EOL> data = [ data ] <EOL> req . callcount = len ( data ) <EOL> for call in data : <EOL> yield ExtCallContext ( req , namespace , call ) <EOL> def extract_path ( self , context ) : <EOL> path = list ( context . namespace ) <EOL> if context . action : <EOL> path . append ( context . action ) <EOL> path . append ( context . method ) <EOL> return path <EOL> def read_std_arguments ( self , context ) : <EOL> funcdef = context . funcdef <EOL> notation = funcdef . extra_options . get ( '<STR_LIT>' , <EOL> self . default_params_notation ) <EOL> args = context . params <EOL> if notation == '<STR_LIT>' : <EOL> kw = dict ( <EOL> ( argdef . name , fromjson ( argdef . datatype , arg ) ) <EOL> for argdef , arg in zip ( funcdef . arguments , args ) <EOL> ) <EOL> elif notation == '<STR_LIT>' : <EOL> if len ( args ) == <NUM_LIT:0> : <EOL> args = [ { } ] <EOL> elif len ( args ) > <NUM_LIT:1> : <EOL> raise ClientSideError ( <EOL> "<STR_LIT>" ) <EOL> args = args [ <NUM_LIT:0> ] <EOL> kw = dict ( <EOL> ( argdef . name , fromjson ( argdef . datatype , args [ argdef . name ] ) ) <EOL> for argdef in funcdef . arguments if argdef . name in args <EOL> ) <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" % notation ) <EOL> return kw <EOL> def read_form_arguments ( self , context ) : <EOL> kw = { } <EOL> for argdef in context . funcdef . arguments : <EOL> value = from_params ( argdef . datatype , context . request . params , <EOL> argdef . name , set ( ) ) <EOL> if value is not Unset : <EOL> kw [ argdef . name ] = value <EOL> return kw <EOL> def read_arguments ( self , context ) : <EOL> if isinstance ( context , ExtCallContext ) : <EOL> kwargs = self . read_std_arguments ( context ) <EOL> elif isinstance ( context , FormExtCallContext ) : <EOL> kwargs = self . read_form_arguments ( context ) <EOL> wsme . runtime . check_arguments ( context . funcdef , ( ) , kwargs ) <EOL> return kwargs <EOL> def encode_result ( self , context , result ) : <EOL> return json . dumps ( { <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : context . tid , <EOL> '<STR_LIT:action>' : context . action , <EOL> '<STR_LIT>' : context . method , <EOL> '<STR_LIT:result>' : tojson ( context . funcdef . return_type , result ) <EOL> } ) <EOL> def encode_error ( self , context , infos ) : <EOL> return json . dumps ( { <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : context . tid , <EOL> '<STR_LIT:action>' : context . action , <EOL> '<STR_LIT>' : context . method , <EOL> '<STR_LIT:message>' : '<STR_LIT>' % infos , <EOL> '<STR_LIT>' : infos [ '<STR_LIT>' ] } ) <EOL> def prepare_response_body ( self , request , results ) : <EOL> r = "<STR_LIT>" . join ( results ) <EOL> if request . wsme_extdirect_batchcall : <EOL> return "<STR_LIT>" % r <EOL> else : <EOL> return r <EOL> def get_response_status ( self , request ) : <EOL> return <NUM_LIT:200> <EOL> def get_response_contenttype ( self , request ) : <EOL> return "<STR_LIT>" <EOL> def fullns ( self , ns ) : <EOL> return ns and '<STR_LIT>' % ( self . namespace , ns ) or self . namespace <EOL> @ expose ( '<STR_LIT>' , "<STR_LIT>" ) <EOL> @ expose ( '<STR_LIT>' , "<STR_LIT>" ) <EOL> def api ( self ) : <EOL> namespaces = { } <EOL> for path , funcdef in self . root . getapi ( ) : <EOL> if len ( path ) > <NUM_LIT:1> : <EOL> namespace = '<STR_LIT:.>' . join ( path [ : - <NUM_LIT:2> ] ) <EOL> action = path [ - <NUM_LIT:2> ] <EOL> else : <EOL> namespace = '<STR_LIT>' <EOL> action = '<STR_LIT>' <EOL> if namespace not in namespaces : <EOL> namespaces [ namespace ] = { } <EOL> if action not in namespaces [ namespace ] : <EOL> namespaces [ namespace ] [ action ] = [ ] <EOL> notation = funcdef . extra_options . get ( '<STR_LIT>' , <EOL> self . default_params_notation ) <EOL> method = { <EOL> '<STR_LIT:name>' : funcdef . name } <EOL> if funcdef . extra_options . get ( '<STR_LIT>' , False ) : <EOL> method [ '<STR_LIT>' ] = True <EOL> method [ '<STR_LIT>' ] = <NUM_LIT:1> if notation == '<STR_LIT>' else len ( funcdef . arguments ) <EOL> namespaces [ namespace ] [ action ] . append ( method ) <EOL> webpath = self . root . _webpath <EOL> if webpath and not webpath . endswith ( '<STR_LIT:/>' ) : <EOL> webpath += '<STR_LIT:/>' <EOL> return APIDefinitionGenerator ( ) . render ( <EOL> namespaces = namespaces , <EOL> webpath = webpath , <EOL> rootns = self . namespace , <EOL> fullns = self . fullns , <EOL> ) <EOL> def encode_sample_value ( self , datatype , value , format = False ) : <EOL> r = tojson ( datatype , value ) <EOL> content = json . dumps ( r , ensure_ascii = False , indent = <NUM_LIT:4> if format else <NUM_LIT:0> , <EOL> sort_keys = format ) <EOL> return ( '<STR_LIT>' , content ) </s>
