<s> import weakref <EOL> import pkg_resources <EOL> from wsme . exc import ClientSideError <EOL> __all__ = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> ] <EOL> registered_protocols = { } <EOL> def _cfg ( f ) : <EOL> cfg = getattr ( f , '<STR_LIT>' , None ) <EOL> if cfg is None : <EOL> f . _cfg = cfg = { } <EOL> return cfg <EOL> class expose ( object ) : <EOL> def __init__ ( self , path , content_type ) : <EOL> self . path = path <EOL> self . content_type = content_type <EOL> def __call__ ( self , func ) : <EOL> func . exposed = True <EOL> cfg = _cfg ( func ) <EOL> cfg [ '<STR_LIT>' ] = self . content_type <EOL> cfg . setdefault ( '<STR_LIT>' , [ ] ) . append ( self . path ) <EOL> return func <EOL> class CallContext ( object ) : <EOL> def __init__ ( self , request ) : <EOL> self . _request = weakref . ref ( request ) <EOL> self . path = None <EOL> self . func = None <EOL> self . funcdef = None <EOL> @ property <EOL> def request ( self ) : <EOL> return self . _request ( ) <EOL> class ObjectDict ( object ) : <EOL> def __init__ ( self , obj ) : <EOL> self . obj = obj <EOL> def __getitem__ ( self , name ) : <EOL> return getattr ( self . obj , name ) <EOL> class Protocol ( object ) : <EOL> name = None <EOL> displayname = None <EOL> content_types = [ ] <EOL> def resolve_path ( self , path ) : <EOL> if '<STR_LIT:$>' in path : <EOL> from string import Template <EOL> s = Template ( path ) <EOL> path = s . substitute ( ObjectDict ( self ) ) <EOL> return path <EOL> def iter_routes ( self ) : <EOL> for attrname in dir ( self ) : <EOL> attr = getattr ( self , attrname ) <EOL> if getattr ( attr , '<STR_LIT>' , False ) : <EOL> for path in _cfg ( attr ) [ '<STR_LIT>' ] : <EOL> yield self . resolve_path ( path ) , attr <EOL> def accept ( self , request ) : <EOL> return request . headers . get ( '<STR_LIT:Content-Type>' ) in self . content_types <EOL> def iter_calls ( self , request ) : <EOL> pass <EOL> def extract_path ( self , context ) : <EOL> pass <EOL> def read_arguments ( self , context ) : <EOL> pass <EOL> def encode_result ( self , context , result ) : <EOL> pass <EOL> def encode_sample_value ( self , datatype , value , format = False ) : <EOL> return ( '<STR_LIT:none>' , '<STR_LIT>' ) <EOL> def encode_sample_params ( self , params , format = False ) : <EOL> return ( '<STR_LIT:none>' , '<STR_LIT>' ) <EOL> def encode_sample_result ( self , datatype , value , format = False ) : <EOL> return ( '<STR_LIT:none>' , '<STR_LIT>' ) <EOL> def register_protocol ( protocol ) : <EOL> registered_protocols [ protocol . name ] = protocol <EOL> def getprotocol ( name , ** options ) : <EOL> protocol_class = registered_protocols . get ( name ) <EOL> if protocol_class is None : <EOL> for entry_point in pkg_resources . iter_entry_points ( <EOL> '<STR_LIT>' , name ) : <EOL> if entry_point . name == name : <EOL> protocol_class = entry_point . load ( ) <EOL> if protocol_class is None : <EOL> raise ValueError ( "<STR_LIT>" % name ) <EOL> registered_protocols [ name ] = protocol_class <EOL> return protocol_class ( ** options ) <EOL> def media_type_accept ( request , content_types ) : <EOL> """<STR_LIT>""" <EOL> if request . method in [ '<STR_LIT:GET>' , '<STR_LIT>' ] : <EOL> if request . accept : <EOL> if request . accept . best_match ( content_types ) : <EOL> return True <EOL> error_message = ( '<STR_LIT>' <EOL> % ( request . accept , content_types ) ) <EOL> raise ClientSideError ( error_message , status_code = <NUM_LIT> ) <EOL> elif request . method in [ '<STR_LIT>' , '<STR_LIT:POST>' , '<STR_LIT>' ] : <EOL> content_type = request . headers . get ( '<STR_LIT:Content-Type>' ) <EOL> if content_type : <EOL> for ct in content_types : <EOL> if request . headers . get ( '<STR_LIT:Content-Type>' , '<STR_LIT>' ) . startswith ( ct ) : <EOL> return True <EOL> error_message = ( '<STR_LIT>' <EOL> % ( content_type , content_types ) ) <EOL> raise ClientSideError ( error_message , status_code = <NUM_LIT> ) <EOL> else : <EOL> raise ClientSideError ( '<STR_LIT>' ) <EOL> elif request . method in [ '<STR_LIT>' ] : <EOL> return True <EOL> return False </s>
