<s> import re <EOL> import logging <EOL> from functools import wraps <EOL> from inspect import getargspec <EOL> from flask . wrappers import Response <EOL> from flask import current_app , request , jsonify <EOL> from flask_jsonrpc . site import jsonrpc_site <EOL> from flask_jsonrpc . _compat import ( b , u , text_type , string_types , <EOL> OrderedDict , NativeStringIO ) <EOL> from flask_jsonrpc . types import ( Object , Number , Boolean , String , Array , <EOL> Nil , Any , Type ) <EOL> from flask_jsonrpc . helpers import ( make_response , jsonify_status_code , <EOL> extract_raw_data_request , authenticate ) <EOL> from flask_jsonrpc . exceptions import ( Error , ParseError , InvalidRequestError , <EOL> MethodNotFoundError , InvalidParamsError , <EOL> ServerError , RequestPostError , <EOL> InvalidCredentialsError , OtherError ) <EOL> default_site = jsonrpc_site <EOL> KWARG_RE = re . compile ( <EOL> r'<STR_LIT>' ) <EOL> SIG_RE = re . compile ( <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' ) <EOL> class JSONRPCTypeCheckingUnavailable ( Exception ) : <EOL> pass <EOL> def _type_checking_available ( sig = '<STR_LIT>' , validate = False ) : <EOL> if not hasattr ( type , '<STR_LIT>' ) and validate : <EOL> raise JSONRPCTypeCheckingUnavailable ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' . format ( sig ) ) <EOL> def _validate_arg ( value , expected ) : <EOL> """<STR_LIT>""" <EOL> if type ( value ) == expected : <EOL> return True <EOL> return False <EOL> def _eval_arg_type ( arg_type , T = Any , arg = None , sig = None ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> T = eval ( arg_type ) <EOL> except Exception as e : <EOL> raise ValueError ( '<STR_LIT>' . format ( arg_type , arg , sig , text_type ( e ) ) ) <EOL> else : <EOL> if type ( T ) not in ( type , Type ) : <EOL> raise TypeError ( '<STR_LIT>' . format ( repr ( T ) , arg , sig ) ) <EOL> return T <EOL> def _parse_sig ( sig , arg_names , validate = False ) : <EOL> """<STR_LIT>""" <EOL> d = SIG_RE . match ( sig ) <EOL> if not d : <EOL> raise ValueError ( '<STR_LIT>' . format ( sig ) ) <EOL> d = d . groupdict ( ) <EOL> ret = [ ( n , Any ) for n in arg_names ] <EOL> if text_type ( '<STR_LIT>' ) in d and type ( d [ '<STR_LIT>' ] ) in string_types and d [ '<STR_LIT>' ] . strip ( ) : <EOL> for i , arg in enumerate ( d [ '<STR_LIT>' ] . strip ( ) . split ( '<STR_LIT:U+002C>' ) ) : <EOL> _type_checking_available ( sig , validate ) <EOL> if text_type ( '<STR_LIT:=>' ) in arg : <EOL> if not type ( ret ) is OrderedDict : <EOL> ret = OrderedDict ( ret ) <EOL> dk = KWARG_RE . match ( arg ) <EOL> if not dk : <EOL> raise ValueError ( '<STR_LIT>' . format ( arg , sig ) ) <EOL> dk = dk . groupdict ( ) <EOL> if not sum ( [ ( k in dk and type ( dk [ k ] ) in string_types and bool ( dk [ k ] . strip ( ) ) ) <EOL> for k in ( '<STR_LIT>' , '<STR_LIT>' ) ] ) : <EOL> raise ValueError ( '<STR_LIT>' . format ( arg , sig ) ) <EOL> ret [ dk [ '<STR_LIT>' ] ] = _eval_arg_type ( dk [ '<STR_LIT>' ] , None , arg , sig ) <EOL> else : <EOL> if type ( ret ) is OrderedDict : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' . format ( sig ) ) <EOL> if len ( ret ) < i + <NUM_LIT:1> : <EOL> ret . append ( ( text_type ( i ) , _eval_arg_type ( arg , None , arg , sig ) ) ) <EOL> else : <EOL> ret [ i ] = ( ret [ i ] [ <NUM_LIT:0> ] , _eval_arg_type ( arg , None , arg , sig ) ) <EOL> if not type ( ret ) is OrderedDict : <EOL> ret = OrderedDict ( ret ) <EOL> return ( d [ '<STR_LIT>' ] , <EOL> ret , <EOL> ( _eval_arg_type ( d [ '<STR_LIT>' ] , Any , '<STR_LIT>' , sig ) <EOL> if d [ '<STR_LIT>' ] else Any ) ) <EOL> def _inject_args ( sig , types ) : <EOL> """<STR_LIT>""" <EOL> if '<STR_LIT:(>' in sig : <EOL> parts = sig . split ( '<STR_LIT:(>' ) <EOL> sig = '<STR_LIT>' . format ( <EOL> parts [ <NUM_LIT:0> ] , '<STR_LIT:U+002CU+0020>' . join ( types ) , <EOL> ( '<STR_LIT:U+002CU+0020>' if parts [ <NUM_LIT:1> ] . index ( '<STR_LIT:)>' ) > <NUM_LIT:0> else '<STR_LIT>' ) , parts [ <NUM_LIT:1> ] <EOL> ) <EOL> else : <EOL> sig = '<STR_LIT>' . format ( sig , '<STR_LIT:U+002CU+0020>' . join ( types ) ) <EOL> return sig <EOL> def _site_api ( site ) : <EOL> def wrapper ( method = '<STR_LIT>' ) : <EOL> response_obj , status_code = site . dispatch ( request , method ) <EOL> if isinstance ( response_obj , Response ) : <EOL> return response_obj , response_obj . status_code <EOL> is_batch = type ( response_obj ) is list <EOL> if current_app . config [ '<STR_LIT>' ] : <EOL> logging . debug ( '<STR_LIT>' , extract_raw_data_request ( request ) ) <EOL> logging . debug ( '<STR_LIT>' , status_code , response_obj ) <EOL> return jsonify_status_code ( status_code , response_obj , is_batch = is_batch ) , status_code <EOL> return wrapper <EOL> class JSONRPC ( object ) : <EOL> def __init__ ( self , app = None , service_url = '<STR_LIT>' , auth_backend = authenticate , site = default_site , <EOL> enable_web_browsable_api = False ) : <EOL> self . service_url = service_url <EOL> self . browse_url = self . _make_browse_url ( service_url ) <EOL> self . enable_web_browsable_api = enable_web_browsable_api <EOL> self . auth_backend = auth_backend <EOL> self . site = site <EOL> self . site_api = _site_api ( site ) <EOL> if app is not None : <EOL> self . app = app <EOL> self . init_app ( self . app ) <EOL> else : <EOL> self . app = None <EOL> def _unique_name ( self , suffix = '<STR_LIT>' ) : <EOL> st = '<STR_LIT:.>' . join ( ( self . service_url + suffix ) . split ( '<STR_LIT:/>' ) ) <EOL> st = st [ <NUM_LIT:1> : : ] if st . startswith ( '<STR_LIT:.>' ) else st <EOL> st = st [ <NUM_LIT:0> : - <NUM_LIT:1> ] if st . endswith ( '<STR_LIT:.>' ) else st <EOL> return st <EOL> def _make_browse_url ( self , service_url ) : <EOL> return service_url + '<STR_LIT>' if not service_url . endswith ( '<STR_LIT:/>' ) else service_url + '<STR_LIT>' <EOL> def _register_browse ( self , app ) : <EOL> if app . config [ '<STR_LIT>' ] or self . enable_web_browsable_api : <EOL> self . _enable_web_browsable_api ( app ) <EOL> def _enable_web_browsable_api ( self , app , url_prefix = None ) : <EOL> from flask_jsonrpc . views import browse <EOL> if url_prefix is None : <EOL> url_prefix = self . browse_url <EOL> self . browse_url = url_prefix <EOL> app . register_blueprint ( browse . mod , url_prefix = url_prefix , <EOL> jsonrpc_site_name = self . _unique_name ( ) , jsonrpc_site = self . site ) <EOL> def init_app ( self , app ) : <EOL> self . _register_browse ( app ) <EOL> app . add_url_rule ( self . service_url , self . _unique_name ( ) , self . site_api , methods = [ '<STR_LIT:POST>' , '<STR_LIT>' ] ) <EOL> app . add_url_rule ( self . service_url + '<STR_LIT>' , self . _unique_name ( '<STR_LIT>' ) , self . site_api , <EOL> methods = [ '<STR_LIT:GET>' , '<STR_LIT>' ] ) <EOL> def register_blueprint ( self , blueprint ) : <EOL> blueprint . add_url_rule ( self . service_url , '<STR_LIT>' , self . site_api , methods = [ '<STR_LIT:POST>' , '<STR_LIT>' ] ) <EOL> blueprint . add_url_rule ( self . service_url + '<STR_LIT>' , '<STR_LIT>' , self . site_api , methods = [ '<STR_LIT:GET>' , '<STR_LIT>' ] ) <EOL> def method ( self , name , authenticated = False , safe = False , validate = False , ** options ) : <EOL> def decorator ( f ) : <EOL> arg_names = getargspec ( f ) [ <NUM_LIT:0> ] <EOL> X = { '<STR_LIT:name>' : name , '<STR_LIT>' : arg_names } <EOL> if authenticated : <EOL> X [ '<STR_LIT>' ] = [ '<STR_LIT:username>' , '<STR_LIT:password>' ] + X [ '<STR_LIT>' ] <EOL> X [ '<STR_LIT:name>' ] = _inject_args ( X [ '<STR_LIT:name>' ] , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> _f = self . auth_backend ( f , authenticated ) <EOL> else : <EOL> _f = f <EOL> method , arg_types , return_type = _parse_sig ( X [ '<STR_LIT:name>' ] , X [ '<STR_LIT>' ] , validate ) <EOL> _f . json_args = X [ '<STR_LIT>' ] <EOL> _f . json_arg_types = arg_types <EOL> _f . json_return_type = return_type <EOL> _f . json_method = method <EOL> _f . json_safe = safe <EOL> _f . json_sig = X [ '<STR_LIT:name>' ] <EOL> _f . json_validate = validate <EOL> self . site . register ( method , _f ) <EOL> return _f <EOL> return decorator </s>
