<s> import datetime , decimal <EOL> import sys <EOL> from functools import wraps <EOL> from uuid import uuid1 <EOL> from jsonrpc . _json import loads , dumps <EOL> from jsonrpc . exceptions import * <EOL> from jsonrpc . _types import * <EOL> from django . conf import settings <EOL> from django . core import signals <EOL> from django . utils . encoding import smart_text <EOL> empty_dec = lambda f : f <EOL> try : <EOL> from django . views . decorators . csrf import csrf_exempt <EOL> except ( NameError , ImportError ) : <EOL> csrf_exempt = empty_dec <EOL> from django . core . serializers . json import DjangoJSONEncoder <EOL> NoneType = type ( None ) <EOL> encode_kw = lambda p : dict ( [ ( str ( k ) , v ) for k , v in p . items ( ) ] ) <EOL> def trim_docstring ( docstring ) : <EOL> if not docstring : <EOL> return '<STR_LIT>' <EOL> lines = docstring . expandtabs ( ) . splitlines ( ) <EOL> indent = sys . maxsize <EOL> for line in lines [ <NUM_LIT:1> : ] : <EOL> stripped = line . lstrip ( ) <EOL> if stripped : <EOL> indent = min ( indent , len ( line ) - len ( stripped ) ) <EOL> trimmed = [ lines [ <NUM_LIT:0> ] . strip ( ) ] <EOL> if indent < sys . maxsize : <EOL> for line in lines [ <NUM_LIT:1> : ] : <EOL> trimmed . append ( line [ indent : ] . rstrip ( ) ) <EOL> while trimmed and not trimmed [ - <NUM_LIT:1> ] : <EOL> trimmed . pop ( ) <EOL> while trimmed and not trimmed [ <NUM_LIT:0> ] : <EOL> trimmed . pop ( <NUM_LIT:0> ) <EOL> return '<STR_LIT:\n>' . join ( trimmed ) <EOL> def encode_kw11 ( p ) : <EOL> if not type ( p ) is dict : <EOL> return { } <EOL> ret = p . copy ( ) <EOL> removes = [ ] <EOL> for k , v in ret . items ( ) : <EOL> try : <EOL> int ( k ) <EOL> except ValueError : <EOL> pass <EOL> else : <EOL> removes . append ( k ) <EOL> for k in removes : <EOL> ret . pop ( k ) <EOL> return ret <EOL> def encode_arg11 ( p ) : <EOL> if type ( p ) is list : <EOL> return p <EOL> elif not type ( p ) is dict : <EOL> return [ ] <EOL> else : <EOL> pos = [ ] <EOL> d = encode_kw ( p ) <EOL> for k , v in d . items ( ) : <EOL> try : <EOL> pos . append ( int ( k ) ) <EOL> except ValueError : <EOL> pass <EOL> pos = list ( set ( pos ) ) <EOL> pos . sort ( ) <EOL> return [ d [ str ( i ) ] for i in pos ] <EOL> def validate_params ( method , D ) : <EOL> if type ( D [ '<STR_LIT>' ] ) == Object : <EOL> keys = list ( method . json_arg_types . keys ( ) ) <EOL> if len ( keys ) != len ( D [ '<STR_LIT>' ] ) : <EOL> raise InvalidParamsError ( '<STR_LIT>' % <EOL> method . json_sig ) <EOL> for k in keys : <EOL> if not k in D [ '<STR_LIT>' ] : <EOL> print ( '<STR_LIT>' , k , D , '<STR_LIT>' ) <EOL> raise InvalidParamsError ( '<STR_LIT>' % <EOL> ( k , method . json_sig ) ) <EOL> if not Any . kind ( D [ '<STR_LIT>' ] [ k ] ) == method . json_arg_types [ k ] : <EOL> raise InvalidParamsError ( <EOL> '<STR_LIT>' % <EOL> ( type ( D [ '<STR_LIT>' ] [ k ] ) , method . json_arg_types [ k ] , <EOL> method . json_sig ) ) <EOL> elif type ( D [ '<STR_LIT>' ] ) == Array : <EOL> arg_types = list ( method . json_arg_types . values ( ) ) <EOL> try : <EOL> for i , arg in enumerate ( D [ '<STR_LIT>' ] ) : <EOL> if not Any . kind ( arg ) == arg_types [ i ] : <EOL> raise InvalidParamsError ( <EOL> '<STR_LIT>' % <EOL> ( type ( arg ) , arg_types [ i ] , method . json_sig ) ) <EOL> except IndexError : <EOL> raise InvalidParamsError ( '<STR_LIT>' % <EOL> method . json_sig ) <EOL> else : <EOL> if len ( D [ '<STR_LIT>' ] ) != len ( arg_types ) : <EOL> raise InvalidParamsError ( '<STR_LIT>' % <EOL> method . json_sig ) <EOL> class JSONRPCSite ( object ) : <EOL> "<STR_LIT>" <EOL> def __init__ ( self , json_encoder = DjangoJSONEncoder ) : <EOL> self . urls = { } <EOL> self . uuid = str ( uuid1 ( ) ) <EOL> self . version = '<STR_LIT:1.0>' <EOL> self . name = '<STR_LIT>' <EOL> self . register ( '<STR_LIT>' , self . describe ) <EOL> self . set_json_encoder ( json_encoder ) <EOL> def set_json_encoder ( self , json_encoder = DjangoJSONEncoder ) : <EOL> self . json_encoder = json_encoder <EOL> def register ( self , name , method ) : <EOL> self . urls [ smart_text ( name ) ] = method <EOL> def empty_response ( self , version = '<STR_LIT:1.0>' ) : <EOL> resp = { '<STR_LIT:id>' : None } <EOL> if version == '<STR_LIT>' : <EOL> resp [ '<STR_LIT:version>' ] = version <EOL> return resp <EOL> if version == '<STR_LIT>' : <EOL> resp [ '<STR_LIT>' ] = version <EOL> resp . update ( { '<STR_LIT:error>' : None , '<STR_LIT:result>' : None } ) <EOL> return resp <EOL> def validate_get ( self , request , method ) : <EOL> encode_get_params = lambda r : dict ( [ ( k , v [ <NUM_LIT:0> ] if len ( v ) == <NUM_LIT:1> else v ) for k , v in r ] ) <EOL> if request . method == '<STR_LIT:GET>' : <EOL> method = smart_text ( method ) <EOL> if method in self . urls and getattr ( self . urls [ method ] , '<STR_LIT>' , <EOL> False ) : <EOL> D = { <EOL> '<STR_LIT>' : encode_get_params ( request . GET . lists ( ) ) , <EOL> '<STR_LIT>' : method , <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT:version>' : '<STR_LIT>' <EOL> } <EOL> return True , D <EOL> return False , { } <EOL> def response_dict ( self , request , D , <EOL> is_batch = False , <EOL> version_hint = '<STR_LIT:1.0>' , <EOL> json_encoder = None ) : <EOL> json_encoder = json_encoder or self . json_encoder <EOL> version = version_hint <EOL> response = self . empty_response ( version = version ) <EOL> apply_version = { <EOL> '<STR_LIT>' : <EOL> lambda f , r , p : f ( r , ** encode_kw ( p ) ) if type ( p ) is dict else f ( r , * p ) , <EOL> '<STR_LIT>' : <EOL> lambda f , r , p : f ( r , * encode_arg11 ( p ) , ** encode_kw ( encode_kw11 ( p ) ) ) , <EOL> '<STR_LIT:1.0>' : lambda f , r , p : f ( r , * p ) <EOL> } <EOL> try : <EOL> if '<STR_LIT>' not in D : <EOL> D [ '<STR_LIT>' ] = [ ] <EOL> if '<STR_LIT>' not in D or '<STR_LIT>' not in D : <EOL> raise InvalidParamsError ( <EOL> '<STR_LIT>' ) <EOL> if D [ '<STR_LIT>' ] not in self . urls : <EOL> raise MethodNotFoundError ( <EOL> '<STR_LIT>' % ( <EOL> '<STR_LIT:\n>' . join ( self . urls . keys ( ) ) ) ) <EOL> if '<STR_LIT>' in D : <EOL> if str ( D [ '<STR_LIT>' ] ) not in apply_version : <EOL> raise InvalidRequestError ( <EOL> '<STR_LIT>' % D [ '<STR_LIT>' ] ) <EOL> version = request . jsonrpc_version = response [ '<STR_LIT>' ] = str ( <EOL> D [ '<STR_LIT>' ] ) <EOL> elif '<STR_LIT:version>' in D : <EOL> if str ( D [ '<STR_LIT:version>' ] ) not in apply_version : <EOL> raise InvalidRequestError ( <EOL> '<STR_LIT>' % D [ '<STR_LIT:version>' ] ) <EOL> version = request . jsonrpc_version = response [ '<STR_LIT:version>' ] = str ( <EOL> D [ '<STR_LIT:version>' ] ) <EOL> else : <EOL> request . jsonrpc_version = '<STR_LIT:1.0>' <EOL> method = self . urls [ str ( D [ '<STR_LIT>' ] ) ] <EOL> if getattr ( method , '<STR_LIT>' , False ) : <EOL> validate_params ( method , D ) <EOL> if '<STR_LIT:id>' in D and D [ '<STR_LIT:id>' ] is not None : <EOL> response [ '<STR_LIT:id>' ] = D [ '<STR_LIT:id>' ] <EOL> if version in ( '<STR_LIT>' , '<STR_LIT>' ) and '<STR_LIT:error>' in response : <EOL> response . pop ( '<STR_LIT:error>' ) <EOL> elif is_batch : <EOL> raise InvalidRequestError <EOL> R = apply_version [ version ] ( method , request , D [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT:id>' not in D or ( '<STR_LIT:id>' in D and D [ '<STR_LIT:id>' ] is None ) : <EOL> return None , <NUM_LIT> <EOL> if isinstance ( R , tuple ) : <EOL> R = list ( R ) <EOL> encoder = json_encoder ( ) <EOL> builtin_types = ( dict , list , set , NoneType , bool , six . text_type <EOL> ) + six . integer_types + six . string_types <EOL> if all ( not isinstance ( R , e ) for e in builtin_types ) : <EOL> try : <EOL> rs = encoder . default ( R ) <EOL> except TypeError as exc : <EOL> raise TypeError ( "<STR_LIT>" % R ) <EOL> response [ '<STR_LIT:result>' ] = R <EOL> status = <NUM_LIT:200> <EOL> except Error as e : <EOL> response [ '<STR_LIT:error>' ] = e . json_rpc_format <EOL> if version in ( '<STR_LIT>' , '<STR_LIT>' ) and '<STR_LIT:result>' in response : <EOL> response . pop ( '<STR_LIT:result>' ) <EOL> status = e . status <EOL> except Exception as e : <EOL> signals . got_request_exception . send ( sender = self . __class__ , <EOL> request = request ) <EOL> if settings . DEBUG : <EOL> other_error = OtherError ( e ) <EOL> else : <EOL> other_error = OtherError ( "<STR_LIT>" ) <EOL> response [ '<STR_LIT:error>' ] = other_error . json_rpc_format <EOL> status = other_error . status <EOL> if version in ( '<STR_LIT>' , '<STR_LIT>' ) and '<STR_LIT:result>' in response : <EOL> response . pop ( '<STR_LIT:result>' ) <EOL> if version in ( '<STR_LIT>' , '<STR_LIT>' <EOL> ) and '<STR_LIT:error>' in response and not response [ '<STR_LIT:error>' ] : <EOL> response . pop ( '<STR_LIT:error>' ) <EOL> return response , status <EOL> @ csrf_exempt <EOL> def dispatch ( self , request , method = '<STR_LIT>' , json_encoder = None ) : <EOL> from django . http import HttpResponse <EOL> json_encoder = json_encoder or self . json_encoder <EOL> try : <EOL> response = self . empty_response ( ) <EOL> if request . method . lower ( ) == '<STR_LIT>' : <EOL> valid , D = self . validate_get ( request , method ) <EOL> if not valid : <EOL> raise InvalidRequestError ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> elif not request . method . lower ( ) == '<STR_LIT>' : <EOL> raise RequestPostError <EOL> else : <EOL> try : <EOL> if hasattr ( request , "<STR_LIT:body>" ) : <EOL> D = loads ( request . body . decode ( '<STR_LIT:utf-8>' ) ) <EOL> else : <EOL> D = loads ( request . raw_post_data . decode ( '<STR_LIT:utf-8>' ) ) <EOL> except : <EOL> raise InvalidRequestError <EOL> if type ( D ) is list : <EOL> response = [ self . response_dict ( request , d , <EOL> is_batch = True , <EOL> json_encoder = json_encoder ) [ <NUM_LIT:0> ] <EOL> for d in D ] <EOL> status = <NUM_LIT:200> <EOL> else : <EOL> response , status = self . response_dict ( <EOL> request , D , <EOL> json_encoder = json_encoder ) <EOL> if response is None and ( not '<STR_LIT:id>' in D or D [ '<STR_LIT:id>' ] is None ) : <EOL> return HttpResponse ( '<STR_LIT>' , status = status ) <EOL> json_rpc = dumps ( response , cls = json_encoder ) <EOL> except Error as e : <EOL> response [ '<STR_LIT:error>' ] = e . json_rpc_format <EOL> status = e . status <EOL> json_rpc = dumps ( response , cls = json_encoder ) <EOL> except Exception as e : <EOL> signals . got_request_exception . send ( sender = self . __class__ , <EOL> request = request ) <EOL> if settings . DEBUG : <EOL> other_error = OtherError ( e ) <EOL> else : <EOL> other_error = OtherError ( "<STR_LIT>" ) <EOL> response [ '<STR_LIT:result>' ] = None <EOL> response [ '<STR_LIT:error>' ] = other_error . json_rpc_format <EOL> status = other_error . status <EOL> json_rpc = dumps ( response , cls = json_encoder ) <EOL> return HttpResponse ( json_rpc , <EOL> status = status , <EOL> content_type = '<STR_LIT>' ) <EOL> def procedure_desc ( self , key ) : <EOL> M = self . urls [ key ] <EOL> return { <EOL> '<STR_LIT:name>' : M . json_method , <EOL> '<STR_LIT>' : trim_docstring ( M . __doc__ ) , <EOL> '<STR_LIT>' : M . json_safe , <EOL> '<STR_LIT>' : [ { '<STR_LIT:type>' : str ( Any . kind ( t ) ) , <EOL> '<STR_LIT:name>' : k } for k , t in M . json_arg_types . items ( ) ] , <EOL> '<STR_LIT>' : { '<STR_LIT:type>' : str ( M . json_return_type ) } <EOL> } <EOL> def service_desc ( self ) : <EOL> return { <EOL> '<STR_LIT>' : '<STR_LIT:1.0>' , <EOL> '<STR_LIT:name>' : self . name , <EOL> '<STR_LIT:id>' : '<STR_LIT>' % str ( self . uuid ) , <EOL> '<STR_LIT>' : trim_docstring ( self . __doc__ ) , <EOL> '<STR_LIT:version>' : self . version , <EOL> '<STR_LIT>' : [ self . procedure_desc ( k ) for k in self . urls . keys ( ) <EOL> if self . urls [ k ] != self . describe ] <EOL> } <EOL> def describe ( self , request ) : <EOL> return self . service_desc ( ) <EOL> jsonrpc_site = JSONRPCSite ( ) </s>
