<s> """<STR_LIT>""" <EOL> import httplib2 <EOL> from functools import wraps <EOL> from urlparse import urljoin <EOL> from flask import request , session , json , redirect <EOL> from werkzeug import url_decode , url_encode , url_quote , parse_options_header , Headers <EOL> import oauth2 <EOL> _etree = None <EOL> def get_etree ( ) : <EOL> """<STR_LIT>""" <EOL> global _etree <EOL> if _etree is None : <EOL> try : <EOL> from lxml import etree as _etree <EOL> except ImportError : <EOL> try : <EOL> from xml . etree import cElementTree as _etree <EOL> except ImportError : <EOL> try : <EOL> from xml . etree import ElementTree as _etree <EOL> except ImportError : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> return _etree <EOL> def parse_response ( resp , content , strict = False ) : <EOL> ct , options = parse_options_header ( resp [ '<STR_LIT>' ] ) <EOL> if ct in ( '<STR_LIT:application/json>' , '<STR_LIT>' ) : <EOL> return json . loads ( content ) <EOL> elif ct in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> charset = options . get ( '<STR_LIT>' , '<STR_LIT:utf-8>' ) <EOL> return get_etree ( ) . fromstring ( content . decode ( charset ) ) <EOL> elif ct != '<STR_LIT>' : <EOL> if strict : <EOL> return content <EOL> charset = options . get ( '<STR_LIT>' , '<STR_LIT:utf-8>' ) <EOL> return url_decode ( content , charset = charset ) . to_dict ( ) <EOL> def add_query ( url , args ) : <EOL> if not args : <EOL> return url <EOL> return url + ( '<STR_LIT:?>' in url and '<STR_LIT:&>' or '<STR_LIT:?>' ) + url_encode ( args ) <EOL> def encode_request_data ( data , format ) : <EOL> if format is None : <EOL> return data , None <EOL> elif format == '<STR_LIT>' : <EOL> return json . dumps ( data or { } ) , '<STR_LIT:application/json>' <EOL> elif format == '<STR_LIT>' : <EOL> return url_encode ( data or { } ) , '<STR_LIT>' <EOL> raise TypeError ( '<STR_LIT>' % format ) <EOL> class OAuthResponse ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , resp , content ) : <EOL> self . headers = Headers ( resp ) <EOL> self . raw_data = content <EOL> self . data = parse_response ( resp , content , strict = True ) <EOL> @ property <EOL> def status ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . headers . get ( '<STR_LIT:status>' , type = int ) <EOL> class OAuthClient ( oauth2 . Client ) : <EOL> def request_new_token ( self , uri , callback = None , params = { } ) : <EOL> if callback is not None : <EOL> params [ '<STR_LIT>' ] = callback <EOL> req = oauth2 . Request . from_consumer_and_token ( <EOL> self . consumer , token = self . token , <EOL> http_method = '<STR_LIT:POST>' , http_url = uri , parameters = params , <EOL> is_form_encoded = True ) <EOL> req . sign_request ( self . method , self . consumer , self . token ) <EOL> body = req . to_postdata ( ) <EOL> headers = { <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : str ( len ( body ) ) <EOL> } <EOL> return httplib2 . Http . request ( self , uri , method = '<STR_LIT:POST>' , <EOL> body = body , headers = headers ) <EOL> class OAuthException ( RuntimeError ) : <EOL> """<STR_LIT>""" <EOL> message = None <EOL> def __init__ ( self , message , data = None ) : <EOL> self . message = message <EOL> self . data = data <EOL> def __str__ ( self ) : <EOL> return self . message . encode ( '<STR_LIT:utf-8>' ) <EOL> def __unicode__ ( self ) : <EOL> return self . message <EOL> class OAuth ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . remote_apps = { } <EOL> def remote_app ( self , name , register = True , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> app = OAuthRemoteApp ( self , name , ** kwargs ) <EOL> if register : <EOL> assert name not in self . remote_apps , '<STR_LIT>' <EOL> self . remote_apps [ name ] = app <EOL> return app <EOL> class OAuthRemoteApp ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , oauth , name , base_url , <EOL> request_token_url , <EOL> access_token_url , authorize_url , <EOL> consumer_key , consumer_secret , <EOL> request_token_params = None , <EOL> access_token_params = None , <EOL> access_token_method = '<STR_LIT:GET>' ) : <EOL> self . oauth = oauth <EOL> self . base_url = base_url <EOL> self . name = name <EOL> self . request_token_url = request_token_url <EOL> self . access_token_url = access_token_url <EOL> self . authorize_url = authorize_url <EOL> self . consumer_key = consumer_key <EOL> self . consumer_secret = consumer_secret <EOL> self . tokengetter_func = None <EOL> self . request_token_params = request_token_params or { } <EOL> self . access_token_params = access_token_params or { } <EOL> self . access_token_method = access_token_method <EOL> self . _consumer = oauth2 . Consumer ( self . consumer_key , <EOL> self . consumer_secret ) <EOL> self . _client = OAuthClient ( self . _consumer ) <EOL> def get ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> kwargs [ '<STR_LIT>' ] = '<STR_LIT:GET>' <EOL> return self . request ( * args , ** kwargs ) <EOL> def post ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> kwargs [ '<STR_LIT>' ] = '<STR_LIT:POST>' <EOL> return self . request ( * args , ** kwargs ) <EOL> def put ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> kwargs [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> return self . request ( * args , ** kwargs ) <EOL> def delete ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> kwargs [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> return self . request ( * args , ** kwargs ) <EOL> def make_client ( self ) : <EOL> """<STR_LIT>""" <EOL> return oauth2 . Client ( self . _consumer , self . get_request_token ( ) ) <EOL> def request ( self , url , data = "<STR_LIT>" , headers = None , format = '<STR_LIT>' , <EOL> method = '<STR_LIT:GET>' , content_type = None ) : <EOL> """<STR_LIT>""" <EOL> headers = dict ( headers or { } ) <EOL> client = self . make_client ( ) <EOL> url = self . expand_url ( url ) <EOL> if method == '<STR_LIT:GET>' : <EOL> assert format == '<STR_LIT>' <EOL> if not data : <EOL> url = add_query ( url , data ) <EOL> data = "<STR_LIT>" <EOL> else : <EOL> if content_type is None : <EOL> data , content_type = encode_request_data ( data , format ) <EOL> if content_type is not None : <EOL> headers [ '<STR_LIT:Content-Type>' ] = content_type <EOL> return OAuthResponse ( * client . request ( url , method = method , <EOL> body = data or '<STR_LIT>' , <EOL> headers = headers ) ) <EOL> def expand_url ( self , url ) : <EOL> return urljoin ( self . base_url , url ) <EOL> def generate_request_token ( self , callback = None ) : <EOL> if callback is not None : <EOL> callback = urljoin ( request . url , callback ) <EOL> resp , content = self . _client . request_new_token ( <EOL> self . expand_url ( self . request_token_url ) , callback , <EOL> self . request_token_params ) <EOL> if resp [ '<STR_LIT:status>' ] != '<STR_LIT>' : <EOL> raise OAuthException ( '<STR_LIT>' ) <EOL> data = parse_response ( resp , content ) <EOL> if data is None : <EOL> raise OAuthException ( '<STR_LIT>' + self . name ) <EOL> tup = ( data [ '<STR_LIT>' ] , data [ '<STR_LIT>' ] ) <EOL> session [ self . name + '<STR_LIT>' ] = tup <EOL> return tup <EOL> def get_request_token ( self ) : <EOL> assert self . tokengetter_func is not None , '<STR_LIT>' <EOL> rv = self . tokengetter_func ( ) <EOL> if rv is None : <EOL> rv = session . get ( self . name + '<STR_LIT>' ) <EOL> if rv is None : <EOL> raise OAuthException ( '<STR_LIT>' ) <EOL> return oauth2 . Token ( * rv ) <EOL> def free_request_token ( self ) : <EOL> session . pop ( self . name + '<STR_LIT>' , None ) <EOL> session . pop ( self . name + '<STR_LIT>' , None ) <EOL> def authorize ( self , callback = None ) : <EOL> """<STR_LIT>""" <EOL> if self . request_token_url : <EOL> token = self . generate_request_token ( callback ) [ <NUM_LIT:0> ] <EOL> url = '<STR_LIT>' % ( self . expand_url ( self . authorize_url ) , <EOL> url_quote ( token ) ) <EOL> else : <EOL> assert callback is not None , '<STR_LIT>' <EOL> params = dict ( self . request_token_params ) <EOL> params [ '<STR_LIT>' ] = callback <EOL> params [ '<STR_LIT>' ] = self . consumer_key <EOL> session [ self . name + '<STR_LIT>' ] = callback <EOL> url = add_query ( self . expand_url ( self . authorize_url ) , params ) <EOL> return redirect ( url ) <EOL> def tokengetter ( self , f ) : <EOL> """<STR_LIT>""" <EOL> self . tokengetter_func = f <EOL> return f <EOL> def handle_oauth1_response ( self ) : <EOL> """<STR_LIT>""" <EOL> client = self . make_client ( ) <EOL> resp , content = client . request ( '<STR_LIT>' % ( <EOL> self . expand_url ( self . access_token_url ) , <EOL> request . args [ '<STR_LIT>' ] <EOL> ) , self . access_token_method ) <EOL> data = parse_response ( resp , content ) <EOL> if resp [ '<STR_LIT:status>' ] != '<STR_LIT>' : <EOL> raise OAuthException ( '<STR_LIT>' + self . name , data ) <EOL> return data <EOL> def handle_oauth2_response ( self ) : <EOL> """<STR_LIT>""" <EOL> remote_args = { <EOL> '<STR_LIT:code>' : request . args . get ( '<STR_LIT:code>' ) , <EOL> '<STR_LIT>' : self . consumer_key , <EOL> '<STR_LIT>' : self . consumer_secret , <EOL> '<STR_LIT>' : session . get ( self . name + '<STR_LIT>' ) <EOL> } <EOL> remote_args . update ( self . access_token_params ) <EOL> if self . access_token_method == '<STR_LIT:POST>' : <EOL> resp , content = self . _client . request ( self . access_token_url , <EOL> self . access_token_method , <EOL> url_encode ( remote_args ) ) <EOL> elif self . access_token_method == '<STR_LIT:GET>' : <EOL> url = add_query ( self . expand_url ( self . access_token_url ) , remote_args ) <EOL> resp , content = self . _client . request ( url , self . access_token_method ) <EOL> else : <EOL> raise OAuthException ( '<STR_LIT>' + <EOL> self . access_token_method ) <EOL> data = parse_response ( resp , content ) <EOL> if resp [ '<STR_LIT:status>' ] != '<STR_LIT>' : <EOL> raise OAuthException ( '<STR_LIT>' + self . name , data ) <EOL> return data <EOL> def handle_unknown_response ( self ) : <EOL> """<STR_LIT>""" <EOL> return None <EOL> def authorized_handler ( self , f ) : <EOL> """<STR_LIT>""" <EOL> @ wraps ( f ) <EOL> def decorated ( * args , ** kwargs ) : <EOL> if '<STR_LIT>' in request . args : <EOL> data = self . handle_oauth1_response ( ) <EOL> elif '<STR_LIT:code>' in request . args : <EOL> data = self . handle_oauth2_response ( ) <EOL> else : <EOL> data = self . handle_unknown_response ( ) <EOL> self . free_request_token ( ) <EOL> return f ( * ( ( data , ) + args ) , ** kwargs ) <EOL> return decorated </s>
