<s> import random <EOL> import sys <EOL> import json <EOL> import functools <EOL> import requests <EOL> class JSONRPCError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> class TransportError ( JSONRPCError ) : <EOL> """<STR_LIT>""" <EOL> class ProtocolError ( JSONRPCError ) : <EOL> """<STR_LIT>""" <EOL> class Server ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , url , session = None , ** requests_kwargs ) : <EOL> self . session = session or requests . Session ( ) <EOL> self . session . headers . update ( { <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> self . request = functools . partial ( self . session . post , url , ** requests_kwargs ) <EOL> def send_request ( self , method_name , is_notification , params ) : <EOL> """<STR_LIT>""" <EOL> request_body = self . serialize ( method_name , params , is_notification ) <EOL> try : <EOL> response = self . request ( data = request_body , timeout = <NUM_LIT:30> ) <EOL> except requests . RequestException as requests_exception : <EOL> raise TransportError ( '<STR_LIT>' % method_name , requests_exception ) <EOL> if response . status_code != requests . codes . ok : <EOL> raise TransportError ( response . status_code ) <EOL> if not is_notification : <EOL> try : <EOL> parsed = response . json ( ) <EOL> except ValueError as value_error : <EOL> raise TransportError ( '<STR_LIT>' , value_error ) <EOL> return self . parse_result ( parsed ) <EOL> @ staticmethod <EOL> def parse_result ( result ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( result , dict ) : <EOL> raise ProtocolError ( '<STR_LIT>' ) <EOL> if result . get ( '<STR_LIT:error>' ) : <EOL> code = result [ '<STR_LIT:error>' ] . get ( '<STR_LIT:code>' , '<STR_LIT>' ) <EOL> message = result [ '<STR_LIT:error>' ] . get ( '<STR_LIT:message>' , '<STR_LIT>' ) <EOL> raise ProtocolError ( code , message , result ) <EOL> elif '<STR_LIT:result>' not in result : <EOL> raise ProtocolError ( '<STR_LIT>' ) <EOL> else : <EOL> return result [ '<STR_LIT:result>' ] <EOL> @ staticmethod <EOL> def dumps ( data ) : <EOL> """<STR_LIT>""" <EOL> return json . dumps ( data ) <EOL> def serialize ( self , method_name , params , is_notification ) : <EOL> """<STR_LIT>""" <EOL> data = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : method_name } <EOL> if params : <EOL> data [ '<STR_LIT>' ] = params <EOL> if not is_notification : <EOL> data [ '<STR_LIT:id>' ] = random . randint ( <NUM_LIT:1> , sys . maxsize ) <EOL> return self . dumps ( data ) <EOL> def __getattr__ ( self , method_name ) : <EOL> return Method ( self . __request , method_name ) <EOL> def __request ( self , method_name , args = None , kwargs = None ) : <EOL> """<STR_LIT>""" <EOL> is_notification = kwargs . pop ( '<STR_LIT>' , False ) <EOL> if args and kwargs : <EOL> raise ProtocolError ( '<STR_LIT>' ) <EOL> return self . send_request ( method_name , is_notification , args or kwargs ) <EOL> class Method ( object ) : <EOL> def __init__ ( self , request_method , method_name ) : <EOL> if method_name . startswith ( "<STR_LIT:_>" ) : <EOL> raise AttributeError ( "<STR_LIT>" % method_name ) <EOL> self . __request_method = request_method <EOL> self . __method_name = method_name <EOL> def __getattr__ ( self , method_name ) : <EOL> if method_name . startswith ( "<STR_LIT:_>" ) : <EOL> raise AttributeError ( "<STR_LIT>" % method_name ) <EOL> return Method ( self . __request_method , "<STR_LIT>" % ( self . __method_name , method_name ) ) <EOL> def __call__ ( self , * args , ** kwargs ) : <EOL> return self . __request_method ( self . __method_name , args , kwargs ) </s>
