<s> """<STR_LIT>""" <EOL> from __future__ import unicode_literals <EOL> import sys <EOL> import requests <EOL> import requests_oauthlib <EOL> import logging <EOL> try : <EOL> from urllib . parse import urlparse , urlunparse <EOL> except ImportError : <EOL> from urlparse import urlparse , urlunparse <EOL> from trovebox . objects . trovebox_object import TroveboxObject <EOL> from . errors import TroveboxError , Trovebox404Error , TroveboxDuplicateError <EOL> from . auth import Auth <EOL> if sys . version < '<STR_LIT:3>' : <EOL> TEXT_TYPE = unicode <EOL> else : <EOL> TEXT_TYPE = str <EOL> DUPLICATE_RESPONSE = { "<STR_LIT:code>" : <NUM_LIT> , <EOL> "<STR_LIT:message>" : "<STR_LIT>" } <EOL> class Http ( object ) : <EOL> """<STR_LIT>""" <EOL> _CONFIG_DEFAULTS = { "<STR_LIT>" : None , <EOL> "<STR_LIT>" : True , <EOL> } <EOL> def __init__ ( self , config_file = None , host = None , <EOL> consumer_key = '<STR_LIT>' , consumer_secret = '<STR_LIT>' , <EOL> token = '<STR_LIT>' , token_secret = '<STR_LIT>' , api_version = None ) : <EOL> self . config = dict ( self . _CONFIG_DEFAULTS ) <EOL> if api_version is not None : <EOL> print ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . config [ "<STR_LIT>" ] = api_version <EOL> self . _logger = logging . getLogger ( "<STR_LIT>" ) <EOL> self . auth = Auth ( config_file , host , <EOL> consumer_key , consumer_secret , <EOL> token , token_secret ) <EOL> self . host = self . auth . host <EOL> self . last_url = None <EOL> self . last_params = None <EOL> self . last_response = None <EOL> def configure ( self , ** kwds ) : <EOL> """<STR_LIT>""" <EOL> for item in kwds : <EOL> self . config [ item ] = kwds [ item ] <EOL> def get ( self , endpoint , process_response = True , ** params ) : <EOL> """<STR_LIT>""" <EOL> params = self . _process_params ( params ) <EOL> url = self . _construct_url ( endpoint ) <EOL> if self . auth . consumer_key : <EOL> auth = requests_oauthlib . OAuth1 ( self . auth . consumer_key , <EOL> self . auth . consumer_secret , <EOL> self . auth . token , <EOL> self . auth . token_secret ) <EOL> else : <EOL> auth = None <EOL> with requests . Session ( ) as session : <EOL> session . verify = self . config [ "<STR_LIT>" ] <EOL> response = session . get ( url , params = params , auth = auth ) <EOL> self . _logger . info ( "<STR_LIT>" ) <EOL> self . _logger . info ( "<STR_LIT>" % url ) <EOL> self . _logger . info ( "<STR_LIT>" ) <EOL> self . _logger . info ( response . text [ : <NUM_LIT:1000> ] ) <EOL> if len ( response . text ) > <NUM_LIT:1000> : <EOL> self . _logger . info ( "<STR_LIT>" ) <EOL> self . last_url = url <EOL> self . last_params = params <EOL> self . last_response = response <EOL> if process_response : <EOL> return self . _process_response ( response ) <EOL> else : <EOL> if <NUM_LIT:200> <= response . status_code < <NUM_LIT> : <EOL> return response . text <EOL> else : <EOL> raise TroveboxError ( "<STR_LIT>" % <EOL> ( response . status_code , response . reason ) ) <EOL> def post ( self , endpoint , process_response = True , files = None , ** params ) : <EOL> """<STR_LIT>""" <EOL> params = self . _process_params ( params ) <EOL> url = self . _construct_url ( endpoint ) <EOL> if not self . auth . consumer_key : <EOL> raise TroveboxError ( "<STR_LIT>" ) <EOL> auth = requests_oauthlib . OAuth1 ( self . auth . consumer_key , <EOL> self . auth . consumer_secret , <EOL> self . auth . token , <EOL> self . auth . token_secret ) <EOL> with requests . Session ( ) as session : <EOL> session . verify = self . config [ "<STR_LIT>" ] <EOL> if files : <EOL> response = session . post ( url , params = params , <EOL> files = files , auth = auth ) <EOL> else : <EOL> response = session . post ( url , data = params , auth = auth ) <EOL> self . _logger . info ( "<STR_LIT>" ) <EOL> self . _logger . info ( "<STR_LIT>" % url ) <EOL> self . _logger . info ( "<STR_LIT>" % repr ( params ) ) <EOL> if files : <EOL> self . _logger . info ( "<STR_LIT>" % repr ( files ) ) <EOL> self . _logger . info ( "<STR_LIT>" ) <EOL> self . _logger . info ( response . text [ : <NUM_LIT:1000> ] ) <EOL> if len ( response . text ) > <NUM_LIT:1000> : <EOL> self . _logger . info ( "<STR_LIT>" ) <EOL> self . last_url = url <EOL> self . last_params = params <EOL> self . last_response = response <EOL> if process_response : <EOL> return self . _process_response ( response ) <EOL> else : <EOL> if <NUM_LIT:200> <= response . status_code < <NUM_LIT> : <EOL> return response . text <EOL> else : <EOL> raise TroveboxError ( "<STR_LIT>" % <EOL> ( response . status_code , response . reason ) ) <EOL> def _construct_url ( self , endpoint ) : <EOL> """<STR_LIT>""" <EOL> parsed_url = urlparse ( self . host ) <EOL> scheme = parsed_url [ <NUM_LIT:0> ] <EOL> host = parsed_url [ <NUM_LIT:1> ] <EOL> if scheme == "<STR_LIT>" : <EOL> scheme = "<STR_LIT:http>" <EOL> host = self . host <EOL> if not endpoint . startswith ( "<STR_LIT:/>" ) : <EOL> endpoint = "<STR_LIT:/>" + endpoint <EOL> if self . config [ "<STR_LIT>" ] is not None : <EOL> endpoint = "<STR_LIT>" % ( self . config [ "<STR_LIT>" ] , endpoint ) <EOL> return urlunparse ( ( scheme , host , endpoint , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def _process_params ( self , params ) : <EOL> """<STR_LIT>""" <EOL> processed_params = { } <EOL> for key , value in params . items ( ) : <EOL> processed_params [ key ] = self . _process_param_value ( value ) <EOL> return processed_params <EOL> def _process_param_value ( self , value ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( value , TroveboxObject ) : <EOL> return str ( value . id ) . encode ( '<STR_LIT:utf-8>' ) <EOL> elif isinstance ( value , TEXT_TYPE ) : <EOL> return value . encode ( "<STR_LIT:utf-8>" ) <EOL> elif isinstance ( value , list ) : <EOL> new_list = list ( value ) <EOL> for i , item in enumerate ( new_list ) : <EOL> new_list [ i ] = self . _process_param_value ( item ) <EOL> return b'<STR_LIT:U+002C>' . join ( new_list ) <EOL> elif isinstance ( value , bool ) : <EOL> return b"<STR_LIT:1>" if value else b"<STR_LIT:0>" <EOL> else : <EOL> return str ( value ) . encode ( "<STR_LIT:utf-8>" ) <EOL> @ staticmethod <EOL> def _process_response ( response ) : <EOL> """<STR_LIT>""" <EOL> if response . status_code == <NUM_LIT> : <EOL> raise Trovebox404Error ( "<STR_LIT>" % <EOL> ( response . status_code , response . reason ) ) <EOL> try : <EOL> json_response = response . json ( ) <EOL> code = json_response [ "<STR_LIT:code>" ] <EOL> message = json_response [ "<STR_LIT:message>" ] <EOL> except ( ValueError , KeyError ) : <EOL> if <NUM_LIT:200> <= response . status_code < <NUM_LIT> : <EOL> raise <EOL> else : <EOL> raise TroveboxError ( "<STR_LIT>" % <EOL> ( response . status_code , response . reason ) ) <EOL> if <NUM_LIT:200> <= code < <NUM_LIT> : <EOL> return json_response <EOL> elif ( code == DUPLICATE_RESPONSE [ "<STR_LIT:code>" ] and <EOL> DUPLICATE_RESPONSE [ "<STR_LIT:message>" ] in message ) : <EOL> raise TroveboxDuplicateError ( "<STR_LIT>" % ( code , message ) ) <EOL> else : <EOL> raise TroveboxError ( "<STR_LIT>" % ( code , message ) ) </s>
