<s> """<STR_LIT>""" <EOL> __author__ = '<STR_LIT>' <EOL> import re <EOL> import urllib <EOL> import urlparse <EOL> try : <EOL> from xml . etree import cElementTree as ElementTree <EOL> except ImportError : <EOL> try : <EOL> import cElementTree as ElementTree <EOL> except ImportError : <EOL> try : <EOL> from xml . etree import ElementTree <EOL> except ImportError : <EOL> from elementtree import ElementTree <EOL> import atom . service <EOL> import gdata <EOL> import atom <EOL> import atom . http_interface <EOL> import atom . token_store <EOL> import gdata . auth <EOL> AUTH_SERVER_HOST = '<STR_LIT>' <EOL> SCOPE_URL_PARAM_NAME = '<STR_LIT>' <EOL> CLIENT_LOGIN_SCOPES = { <EOL> '<STR_LIT>' : [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ <EOL> '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ <EOL> '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ <EOL> '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ <EOL> '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ <EOL> '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ <EOL> '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] } <EOL> def lookup_scopes ( service_name ) : <EOL> """<STR_LIT>""" <EOL> if service_name in CLIENT_LOGIN_SCOPES : <EOL> return CLIENT_LOGIN_SCOPES [ service_name ] <EOL> return None <EOL> http_request_handler = atom . service <EOL> class Error ( Exception ) : <EOL> pass <EOL> class CaptchaRequired ( Error ) : <EOL> pass <EOL> class BadAuthentication ( Error ) : <EOL> pass <EOL> class NotAuthenticated ( Error ) : <EOL> pass <EOL> class NonAuthSubToken ( Error ) : <EOL> pass <EOL> class RequestError ( Error ) : <EOL> pass <EOL> class UnexpectedReturnType ( Error ) : <EOL> pass <EOL> class BadAuthenticationServiceURL ( Error ) : <EOL> pass <EOL> class TokenUpgradeFailed ( Error ) : <EOL> pass <EOL> class AuthorizationRequired ( Error ) : <EOL> pass <EOL> class TokenHadNoScope ( Error ) : <EOL> pass <EOL> class GDataService ( atom . service . AtomService ) : <EOL> """<STR_LIT>""" <EOL> handler = None <EOL> auth_token = None <EOL> tokens = None <EOL> def __init__ ( self , email = None , password = None , account_type = '<STR_LIT>' , <EOL> service = None , auth_service_url = None , source = None , server = None , <EOL> additional_headers = None , handler = None , tokens = None , <EOL> http_client = None , token_store = None ) : <EOL> """<STR_LIT>""" <EOL> atom . service . AtomService . __init__ ( self , http_client = http_client , <EOL> token_store = token_store ) <EOL> self . email = email <EOL> self . password = password <EOL> self . account_type = account_type <EOL> self . service = service <EOL> self . auth_service_url = auth_service_url <EOL> self . server = server <EOL> self . additional_headers = additional_headers or { } <EOL> self . __SetSource ( source ) <EOL> self . __captcha_token = None <EOL> self . __captcha_url = None <EOL> self . __gsessionid = None <EOL> if http_request_handler . __name__ == '<STR_LIT>' : <EOL> import gdata . alt . appengine <EOL> self . http_client = gdata . alt . appengine . AppEngineHttpClient ( ) <EOL> def _SetAuthSubToken ( self , auth_token , scopes = None ) : <EOL> """<STR_LIT>""" <EOL> self . SetAuthSubToken ( auth_token , scopes = scopes ) <EOL> def __SetAuthSubToken ( self , auth_token , scopes = None ) : <EOL> """<STR_LIT>""" <EOL> self . _SetAuthSubToken ( auth_token , scopes = scopes ) <EOL> def _GetAuthToken ( self ) : <EOL> """<STR_LIT>""" <EOL> current_scopes = lookup_scopes ( self . service ) <EOL> if current_scopes : <EOL> token = self . token_store . find_token ( current_scopes [ <NUM_LIT:0> ] ) <EOL> if hasattr ( token , '<STR_LIT>' ) : <EOL> return token . auth_header <EOL> return None <EOL> def _GetCaptchaToken ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . __captcha_token <EOL> def __GetCaptchaToken ( self ) : <EOL> return self . _GetCaptchaToken ( ) <EOL> captcha_token = property ( __GetCaptchaToken , <EOL> doc = """<STR_LIT>""" ) <EOL> def _GetCaptchaURL ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . __captcha_url <EOL> def __GetCaptchaURL ( self ) : <EOL> return self . _GetCaptchaURL ( ) <EOL> captcha_url = property ( __GetCaptchaURL , <EOL> doc = """<STR_LIT>""" ) <EOL> def GetAuthSubToken ( self ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( self . current_token , gdata . auth . AuthSubToken ) : <EOL> return self . current_token . get_token_string ( ) <EOL> current_scopes = lookup_scopes ( self . service ) <EOL> if current_scopes : <EOL> token = self . token_store . find_token ( current_scopes [ <NUM_LIT:0> ] ) <EOL> if isinstance ( token , gdata . auth . AuthSubToken ) : <EOL> return token . get_token_string ( ) <EOL> else : <EOL> token = self . token_store . find_token ( atom . token_store . SCOPE_ALL ) <EOL> if isinstance ( token , gdata . auth . ClientLoginToken ) : <EOL> return token . get_token_string ( ) <EOL> return None <EOL> def SetAuthSubToken ( self , token , scopes = None ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( token , gdata . auth . AuthSubToken ) : <EOL> token_string = token <EOL> token = gdata . auth . AuthSubToken ( ) <EOL> token . set_token_string ( token_string ) <EOL> if not token . scopes : <EOL> if scopes is None : <EOL> scopes = lookup_scopes ( self . service ) <EOL> if scopes is None : <EOL> scopes = [ atom . token_store . SCOPE_ALL ] <EOL> token . scopes = scopes <EOL> if self . auto_set_current_token : <EOL> self . current_token = token <EOL> if self . auto_store_tokens : <EOL> self . token_store . add_token ( token ) <EOL> def GetClientLoginToken ( self ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( self . current_token , gdata . auth . ClientLoginToken ) : <EOL> return self . current_token . get_token_string ( ) <EOL> current_scopes = lookup_scopes ( self . service ) <EOL> if current_scopes : <EOL> token = self . token_store . find_token ( current_scopes [ <NUM_LIT:0> ] ) <EOL> if isinstance ( token , gdata . auth . ClientLoginToken ) : <EOL> return token . get_token_string ( ) <EOL> else : <EOL> token = self . token_store . find_token ( atom . token_store . SCOPE_ALL ) <EOL> if isinstance ( token , gdata . auth . ClientLoginToken ) : <EOL> return token . get_token_string ( ) <EOL> return None <EOL> def SetClientLoginToken ( self , token , scopes = None ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( token , gdata . auth . ClientLoginToken ) : <EOL> token_string = token <EOL> token = gdata . auth . ClientLoginToken ( ) <EOL> token . set_token_string ( token_string ) <EOL> if not token . scopes : <EOL> if scopes is None : <EOL> scopes = lookup_scopes ( self . service ) <EOL> if scopes is None : <EOL> scopes = [ atom . token_store . SCOPE_ALL ] <EOL> token . scopes = scopes <EOL> if self . auto_set_current_token : <EOL> self . current_token = token <EOL> if self . auto_store_tokens : <EOL> self . token_store . add_token ( token ) <EOL> def __GetSource ( self ) : <EOL> return self . __source <EOL> def __SetSource ( self , new_source ) : <EOL> self . __source = new_source <EOL> self . additional_headers [ '<STR_LIT>' ] = atom . http_interface . USER_AGENT % ( <EOL> self . __source , ) <EOL> source = property ( __GetSource , __SetSource , <EOL> doc = """<STR_LIT>""" ) <EOL> def ProgrammaticLogin ( self , captcha_token = None , captcha_response = None ) : <EOL> """<STR_LIT>""" <EOL> request_body = gdata . auth . generate_client_login_request_body ( self . email , <EOL> self . password , self . service , self . source , self . account_type , <EOL> captcha_token , captcha_response ) <EOL> if not self . auth_service_url : <EOL> auth_request_url = AUTH_SERVER_HOST + '<STR_LIT>' <EOL> else : <EOL> auth_request_url = self . auth_service_url <EOL> auth_response = self . http_client . request ( '<STR_LIT:POST>' , auth_request_url , <EOL> data = request_body , <EOL> headers = { '<STR_LIT:Content-Type>' : '<STR_LIT>' } ) <EOL> response_body = auth_response . read ( ) <EOL> if auth_response . status == <NUM_LIT:200> : <EOL> self . SetClientLoginToken ( <EOL> gdata . auth . get_client_login_token ( response_body ) ) <EOL> self . __captcha_token = None <EOL> self . __captcha_url = None <EOL> elif auth_response . status == <NUM_LIT> : <EOL> captcha_parameters = gdata . auth . get_captcha_challenge ( response_body , <EOL> captcha_base_url = '<STR_LIT>' % AUTH_SERVER_HOST ) <EOL> if captcha_parameters : <EOL> self . __captcha_token = captcha_parameters [ '<STR_LIT>' ] <EOL> self . __captcha_url = captcha_parameters [ '<STR_LIT:url>' ] <EOL> raise CaptchaRequired , '<STR_LIT>' <EOL> elif response_body . splitlines ( ) [ <NUM_LIT:0> ] == '<STR_LIT>' : <EOL> self . __captcha_token = None <EOL> self . __captcha_url = None <EOL> raise BadAuthentication , '<STR_LIT>' <EOL> else : <EOL> self . __captcha_token = None <EOL> self . __captcha_url = None <EOL> raise Error , '<STR_LIT>' <EOL> elif auth_response . status == <NUM_LIT> : <EOL> self . __captcha_token = None <EOL> self . __captcha_url = None <EOL> raise BadAuthenticationServiceURL , '<STR_LIT>' <EOL> def ClientLogin ( self , username , password , account_type = None , service = None , <EOL> auth_service_url = None , source = None , captcha_token = None , <EOL> captcha_response = None ) : <EOL> """<STR_LIT>""" <EOL> self . email = username <EOL> self . password = password <EOL> if account_type : <EOL> self . account_type = account_type <EOL> if service : <EOL> self . service = service <EOL> if source : <EOL> self . source = source <EOL> if auth_service_url : <EOL> self . auth_service_url = auth_service_url <EOL> self . ProgrammaticLogin ( captcha_token , captcha_response ) <EOL> def GenerateAuthSubURL ( self , next , scope , secure = False , session = True , <EOL> domain = '<STR_LIT:default>' ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( scope , ( list , tuple ) ) : <EOL> scope = ( scope , ) <EOL> return gdata . auth . generate_auth_sub_url ( next , scope , secure = secure , <EOL> session = session , <EOL> request_url = '<STR_LIT>' % AUTH_SERVER_HOST , <EOL> domain = domain ) <EOL> def UpgradeToSessionToken ( self , token = None ) : <EOL> """<STR_LIT>""" <EOL> if token is None : <EOL> scopes = lookup_scopes ( self . service ) <EOL> if scopes : <EOL> token = self . token_store . find_token ( scopes [ <NUM_LIT:0> ] ) <EOL> else : <EOL> token = self . token_store . find_token ( atom . token_store . SCOPE_ALL ) <EOL> if not isinstance ( token , gdata . auth . AuthSubToken ) : <EOL> raise NonAuthSubToken <EOL> self . SetAuthSubToken ( self . upgrade_to_session_token ( token ) ) <EOL> def upgrade_to_session_token ( self , token ) : <EOL> """<STR_LIT>""" <EOL> response = token . perform_request ( self . http_client , '<STR_LIT:GET>' , <EOL> AUTH_SERVER_HOST + '<STR_LIT>' , <EOL> headers = { '<STR_LIT:Content-Type>' : '<STR_LIT>' } ) <EOL> response_body = response . read ( ) <EOL> if response . status == <NUM_LIT:200> : <EOL> token . set_token_string ( <EOL> gdata . auth . token_from_http_body ( response_body ) ) <EOL> return token <EOL> else : <EOL> raise TokenUpgradeFailed ( { '<STR_LIT:status>' : response . status , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:body>' : response_body } ) <EOL> def RevokeAuthSubToken ( self ) : <EOL> """<STR_LIT>""" <EOL> scopes = lookup_scopes ( self . service ) <EOL> token = self . token_store . find_token ( scopes ) <EOL> if not isinstance ( token , gdata . auth . AuthSubToken ) : <EOL> raise NonAuthSubToken <EOL> response = token . perform_request ( self . http_client , '<STR_LIT:GET>' , <EOL> AUTH_SERVER_HOST + '<STR_LIT>' , <EOL> headers = { '<STR_LIT:Content-Type>' : '<STR_LIT>' } ) <EOL> if response . status == <NUM_LIT:200> : <EOL> self . token_store . remove_token ( token ) <EOL> def Get ( self , uri , extra_headers = None , redirects_remaining = <NUM_LIT:4> , <EOL> encoding = '<STR_LIT>' , converter = None ) : <EOL> """<STR_LIT>""" <EOL> if extra_headers is None : <EOL> extra_headers = { } <EOL> if self . __gsessionid is not None : <EOL> if uri . find ( '<STR_LIT>' ) < <NUM_LIT:0> : <EOL> if uri . find ( '<STR_LIT:?>' ) > - <NUM_LIT:1> : <EOL> uri += '<STR_LIT>' % ( self . __gsessionid , ) <EOL> else : <EOL> uri += '<STR_LIT>' % ( self . __gsessionid , ) <EOL> server_response = self . request ( '<STR_LIT:GET>' , uri , <EOL> headers = extra_headers ) <EOL> result_body = server_response . read ( ) <EOL> if server_response . status == <NUM_LIT:200> : <EOL> if converter : <EOL> return converter ( result_body ) <EOL> feed = gdata . GDataFeedFromString ( result_body ) <EOL> if not feed : <EOL> entry = gdata . GDataEntryFromString ( result_body ) <EOL> if not entry : <EOL> return result_body <EOL> return entry <EOL> return feed <EOL> elif server_response . status == <NUM_LIT> : <EOL> if redirects_remaining > <NUM_LIT:0> : <EOL> location = server_response . getheader ( '<STR_LIT>' ) <EOL> if location is not None : <EOL> m = re . compile ( '<STR_LIT>' ) . search ( location ) <EOL> if m is not None : <EOL> self . __gsessionid = m . group ( <NUM_LIT:1> ) <EOL> return GDataService . Get ( self , location , extra_headers , redirects_remaining - <NUM_LIT:1> , <EOL> encoding = encoding , converter = converter ) <EOL> else : <EOL> raise RequestError , { '<STR_LIT:status>' : server_response . status , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:body>' : result_body } <EOL> else : <EOL> raise RequestError , { '<STR_LIT:status>' : server_response . status , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:body>' : result_body } <EOL> else : <EOL> raise RequestError , { '<STR_LIT:status>' : server_response . status , <EOL> '<STR_LIT>' : server_response . reason , '<STR_LIT:body>' : result_body } <EOL> def GetMedia ( self , uri , extra_headers = None ) : <EOL> """<STR_LIT>""" <EOL> response_handle = self . request ( '<STR_LIT:GET>' , uri , <EOL> headers = extra_headers ) <EOL> return gdata . MediaSource ( response_handle , response_handle . getheader ( <EOL> '<STR_LIT:Content-Type>' ) , <EOL> response_handle . getheader ( '<STR_LIT>' ) ) <EOL> def GetEntry ( self , uri , extra_headers = None ) : <EOL> """<STR_LIT>""" <EOL> result = GDataService . Get ( self , uri , extra_headers , <EOL> converter = atom . EntryFromString ) <EOL> if isinstance ( result , atom . Entry ) : <EOL> return result <EOL> else : <EOL> raise UnexpectedReturnType , '<STR_LIT>' <EOL> def GetFeed ( self , uri , extra_headers = None , <EOL> converter = gdata . GDataFeedFromString ) : <EOL> """<STR_LIT>""" <EOL> result = GDataService . Get ( self , uri , extra_headers , converter = converter ) <EOL> if isinstance ( result , atom . Feed ) : <EOL> return result <EOL> else : <EOL> raise UnexpectedReturnType , '<STR_LIT>' <EOL> def GetNext ( self , feed ) : <EOL> """<STR_LIT>""" <EOL> next_link = feed . GetNextLink ( ) <EOL> def ConvertToFeedClass ( xml_string ) : <EOL> return atom . CreateClassFromXMLString ( feed . __class__ , xml_string ) <EOL> if next_link and next_link . href : <EOL> return GDataService . Get ( self , next_link . href , <EOL> converter = ConvertToFeedClass ) <EOL> else : <EOL> return None <EOL> def Post ( self , data , uri , extra_headers = None , url_params = None , <EOL> escape_params = True , redirects_remaining = <NUM_LIT:4> , media_source = None , <EOL> converter = None ) : <EOL> """<STR_LIT>""" <EOL> return GDataService . PostOrPut ( self , '<STR_LIT:POST>' , data , uri , <EOL> extra_headers = extra_headers , url_params = url_params , <EOL> escape_params = escape_params , redirects_remaining = redirects_remaining , <EOL> media_source = media_source , converter = converter ) <EOL> def PostOrPut ( self , verb , data , uri , extra_headers = None , url_params = None , <EOL> escape_params = True , redirects_remaining = <NUM_LIT:4> , media_source = None , <EOL> converter = None ) : <EOL> """<STR_LIT>""" <EOL> if extra_headers is None : <EOL> extra_headers = { } <EOL> if self . __gsessionid is not None : <EOL> if uri . find ( '<STR_LIT>' ) < <NUM_LIT:0> : <EOL> if uri . find ( '<STR_LIT:?>' ) > - <NUM_LIT:1> : <EOL> uri += '<STR_LIT>' % ( self . __gsessionid , ) <EOL> else : <EOL> uri += '<STR_LIT>' % ( self . __gsessionid , ) <EOL> if data and media_source : <EOL> if ElementTree . iselement ( data ) : <EOL> data_str = ElementTree . tostring ( data ) <EOL> else : <EOL> data_str = str ( data ) <EOL> multipart = [ ] <EOL> multipart . append ( '<STR_LIT>' + '<STR_LIT>' ) <EOL> multipart . append ( '<STR_LIT>' + media_source . content_type + '<STR_LIT>' ) <EOL> multipart . append ( '<STR_LIT>' ) <EOL> extra_headers [ '<STR_LIT>' ] = '<STR_LIT:1.0>' <EOL> extra_headers [ '<STR_LIT>' ] = str ( len ( multipart [ <NUM_LIT:0> ] ) + <EOL> len ( multipart [ <NUM_LIT:1> ] ) + len ( multipart [ <NUM_LIT:2> ] ) + <EOL> len ( data_str ) + media_source . content_length ) <EOL> extra_headers [ '<STR_LIT:Content-Type>' ] = '<STR_LIT>' <EOL> server_response = self . request ( verb , uri , <EOL> data = [ multipart [ <NUM_LIT:0> ] , data_str , multipart [ <NUM_LIT:1> ] , media_source . file_handle , <EOL> multipart [ <NUM_LIT:2> ] ] , headers = extra_headers ) <EOL> result_body = server_response . read ( ) <EOL> elif media_source or isinstance ( data , gdata . MediaSource ) : <EOL> if isinstance ( data , gdata . MediaSource ) : <EOL> media_source = data <EOL> extra_headers [ '<STR_LIT>' ] = str ( media_source . content_length ) <EOL> extra_headers [ '<STR_LIT:Content-Type>' ] = media_source . content_type <EOL> server_response = self . request ( verb , uri , <EOL> data = media_source . file_handle , headers = extra_headers ) <EOL> result_body = server_response . read ( ) <EOL> else : <EOL> http_data = data <EOL> content_type = '<STR_LIT>' <EOL> extra_headers [ '<STR_LIT:Content-Type>' ] = content_type <EOL> server_response = self . request ( verb , uri , data = http_data , <EOL> headers = extra_headers ) <EOL> result_body = server_response . read ( ) <EOL> if server_response . status == <NUM_LIT> or server_response . status == <NUM_LIT:200> : <EOL> if converter : <EOL> return converter ( result_body ) <EOL> feed = gdata . GDataFeedFromString ( result_body ) <EOL> if not feed : <EOL> entry = gdata . GDataEntryFromString ( result_body ) <EOL> if not entry : <EOL> return result_body <EOL> return entry <EOL> return feed <EOL> elif server_response . status == <NUM_LIT> : <EOL> if redirects_remaining > <NUM_LIT:0> : <EOL> location = server_response . getheader ( '<STR_LIT>' ) <EOL> if location is not None : <EOL> m = re . compile ( '<STR_LIT>' ) . search ( location ) <EOL> if m is not None : <EOL> self . __gsessionid = m . group ( <NUM_LIT:1> ) <EOL> return GDataService . PostOrPut ( self , verb , data , location , <EOL> extra_headers , url_params , escape_params , <EOL> redirects_remaining - <NUM_LIT:1> , media_source , converter = converter ) <EOL> else : <EOL> raise RequestError , { '<STR_LIT:status>' : server_response . status , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:body>' : result_body } <EOL> else : <EOL> raise RequestError , { '<STR_LIT:status>' : server_response . status , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:body>' : result_body } <EOL> else : <EOL> raise RequestError , { '<STR_LIT:status>' : server_response . status , <EOL> '<STR_LIT>' : server_response . reason , '<STR_LIT:body>' : result_body } <EOL> def Put ( self , data , uri , extra_headers = None , url_params = None , <EOL> escape_params = True , redirects_remaining = <NUM_LIT:3> , media_source = None , <EOL> converter = None ) : <EOL> """<STR_LIT>""" <EOL> return GDataService . PostOrPut ( self , '<STR_LIT>' , data , uri , <EOL> extra_headers = extra_headers , url_params = url_params , <EOL> escape_params = escape_params , redirects_remaining = redirects_remaining , <EOL> media_source = media_source , converter = converter ) <EOL> def Delete ( self , uri , extra_headers = None , url_params = None , <EOL> escape_params = True , redirects_remaining = <NUM_LIT:4> ) : <EOL> """<STR_LIT>""" <EOL> if extra_headers is None : <EOL> extra_headers = { } <EOL> if self . __gsessionid is not None : <EOL> if uri . find ( '<STR_LIT>' ) < <NUM_LIT:0> : <EOL> if uri . find ( '<STR_LIT:?>' ) > - <NUM_LIT:1> : <EOL> uri += '<STR_LIT>' % ( self . __gsessionid , ) <EOL> else : <EOL> uri += '<STR_LIT>' % ( self . __gsessionid , ) <EOL> server_response = self . request ( '<STR_LIT>' , uri , <EOL> headers = extra_headers ) <EOL> result_body = server_response . read ( ) <EOL> if server_response . status == <NUM_LIT:200> : <EOL> return True <EOL> elif server_response . status == <NUM_LIT> : <EOL> if redirects_remaining > <NUM_LIT:0> : <EOL> location = server_response . getheader ( '<STR_LIT>' ) <EOL> if location is not None : <EOL> m = re . compile ( '<STR_LIT>' ) . search ( location ) <EOL> if m is not None : <EOL> self . __gsessionid = m . group ( <NUM_LIT:1> ) <EOL> return GDataService . Delete ( self , location , extra_headers , <EOL> url_params , escape_params , redirects_remaining - <NUM_LIT:1> ) <EOL> else : <EOL> raise RequestError , { '<STR_LIT:status>' : server_response . status , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:body>' : result_body } <EOL> else : <EOL> raise RequestError , { '<STR_LIT:status>' : server_response . status , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:body>' : result_body } <EOL> else : <EOL> raise RequestError , { '<STR_LIT:status>' : server_response . status , <EOL> '<STR_LIT>' : server_response . reason , '<STR_LIT:body>' : result_body } <EOL> def ExtractToken ( url , scopes_included_in_next = True ) : <EOL> """<STR_LIT>""" <EOL> parsed = urlparse . urlparse ( url ) <EOL> token = gdata . auth . AuthSubTokenFromUrl ( parsed [ <NUM_LIT:4> ] ) <EOL> scopes = '<STR_LIT>' <EOL> if scopes_included_in_next : <EOL> for pair in parsed [ <NUM_LIT:4> ] . split ( '<STR_LIT:&>' ) : <EOL> if pair . startswith ( '<STR_LIT>' % SCOPE_URL_PARAM_NAME ) : <EOL> scopes = urllib . unquote_plus ( pair . split ( '<STR_LIT:=>' ) [ <NUM_LIT:1> ] ) <EOL> return ( token , scopes . split ( '<STR_LIT:U+0020>' ) ) <EOL> def GenerateAuthSubRequestUrl ( next , scopes , hd = '<STR_LIT:default>' , secure = False , <EOL> session = True , request_url = '<STR_LIT>' , <EOL> include_scopes_in_next = True ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( scopes , list ) : <EOL> scope = '<STR_LIT:U+0020>' . join ( scopes ) <EOL> else : <EOL> scope = scopes <EOL> if include_scopes_in_next : <EOL> if next . find ( '<STR_LIT:?>' ) > - <NUM_LIT:1> : <EOL> next += '<STR_LIT>' % urllib . urlencode ( { SCOPE_URL_PARAM_NAME : scope } ) <EOL> else : <EOL> next += '<STR_LIT>' % urllib . urlencode ( { SCOPE_URL_PARAM_NAME : scope } ) <EOL> return gdata . auth . GenerateAuthSubUrl ( next = next , scope = scope , secure = secure , <EOL> session = session , request_url = request_url , domain = hd ) <EOL> class Query ( dict ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , feed = None , text_query = None , params = None , <EOL> categories = None ) : <EOL> """<STR_LIT>""" <EOL> self . feed = feed <EOL> self . categories = [ ] <EOL> if text_query : <EOL> self . text_query = text_query <EOL> if isinstance ( params , dict ) : <EOL> for param in params : <EOL> self [ param ] = params [ param ] <EOL> if isinstance ( categories , list ) : <EOL> for category in categories : <EOL> self . categories . append ( category ) <EOL> def _GetTextQuery ( self ) : <EOL> if '<STR_LIT:q>' in self . keys ( ) : <EOL> return self [ '<STR_LIT:q>' ] <EOL> else : <EOL> return None <EOL> def _SetTextQuery ( self , query ) : <EOL> self [ '<STR_LIT:q>' ] = query <EOL> text_query = property ( _GetTextQuery , _SetTextQuery , <EOL> doc = """<STR_LIT>""" ) <EOL> def _GetAuthor ( self ) : <EOL> if '<STR_LIT>' in self . keys ( ) : <EOL> return self [ '<STR_LIT>' ] <EOL> else : <EOL> return None <EOL> def _SetAuthor ( self , query ) : <EOL> self [ '<STR_LIT>' ] = query <EOL> author = property ( _GetAuthor , _SetAuthor , <EOL> doc = """<STR_LIT>""" ) <EOL> def _GetAlt ( self ) : <EOL> if '<STR_LIT>' in self . keys ( ) : <EOL> return self [ '<STR_LIT>' ] <EOL> else : <EOL> return None <EOL> def _SetAlt ( self , query ) : <EOL> self [ '<STR_LIT>' ] = query <EOL> alt = property ( _GetAlt , _SetAlt , <EOL> doc = """<STR_LIT>""" ) <EOL> def _GetUpdatedMin ( self ) : <EOL> if '<STR_LIT>' in self . keys ( ) : <EOL> return self [ '<STR_LIT>' ] <EOL> else : <EOL> return None <EOL> def _SetUpdatedMin ( self , query ) : <EOL> self [ '<STR_LIT>' ] = query <EOL> updated_min = property ( _GetUpdatedMin , _SetUpdatedMin , <EOL> doc = """<STR_LIT>""" ) <EOL> def _GetUpdatedMax ( self ) : <EOL> if '<STR_LIT>' in self . keys ( ) : <EOL> return self [ '<STR_LIT>' ] <EOL> else : <EOL> return None <EOL> def _SetUpdatedMax ( self , query ) : <EOL> self [ '<STR_LIT>' ] = query <EOL> updated_max = property ( _GetUpdatedMax , _SetUpdatedMax , <EOL> doc = """<STR_LIT>""" ) <EOL> def _GetPublishedMin ( self ) : <EOL> if '<STR_LIT>' in self . keys ( ) : <EOL> return self [ '<STR_LIT>' ] <EOL> else : <EOL> return None <EOL> def _SetPublishedMin ( self , query ) : <EOL> self [ '<STR_LIT>' ] = query <EOL> published_min = property ( _GetPublishedMin , _SetPublishedMin , <EOL> doc = """<STR_LIT>""" ) <EOL> def _GetPublishedMax ( self ) : <EOL> if '<STR_LIT>' in self . keys ( ) : <EOL> return self [ '<STR_LIT>' ] <EOL> else : <EOL> return None <EOL> def _SetPublishedMax ( self , query ) : <EOL> self [ '<STR_LIT>' ] = query <EOL> published_max = property ( _GetPublishedMax , _SetPublishedMax , <EOL> doc = """<STR_LIT>""" ) <EOL> def _GetStartIndex ( self ) : <EOL> if '<STR_LIT>' in self . keys ( ) : <EOL> return self [ '<STR_LIT>' ] <EOL> else : <EOL> return None <EOL> def _SetStartIndex ( self , query ) : <EOL> if not isinstance ( query , str ) : <EOL> query = str ( query ) <EOL> self [ '<STR_LIT>' ] = query <EOL> start_index = property ( _GetStartIndex , _SetStartIndex , <EOL> doc = """<STR_LIT>""" ) <EOL> def _GetMaxResults ( self ) : <EOL> if '<STR_LIT>' in self . keys ( ) : <EOL> return self [ '<STR_LIT>' ] <EOL> else : <EOL> return None <EOL> def _SetMaxResults ( self , query ) : <EOL> if not isinstance ( query , str ) : <EOL> query = str ( query ) <EOL> self [ '<STR_LIT>' ] = query <EOL> max_results = property ( _GetMaxResults , _SetMaxResults , <EOL> doc = """<STR_LIT>""" ) <EOL> def _GetOrderBy ( self ) : <EOL> if '<STR_LIT>' in self . keys ( ) : <EOL> return self [ '<STR_LIT>' ] <EOL> else : <EOL> return None <EOL> def _SetOrderBy ( self , query ) : <EOL> self [ '<STR_LIT>' ] = query <EOL> orderby = property ( _GetOrderBy , _SetOrderBy , <EOL> doc = """<STR_LIT>""" ) <EOL> def ToUri ( self ) : <EOL> q_feed = self . feed or '<STR_LIT>' <EOL> category_string = '<STR_LIT:/>' . join ( <EOL> [ urllib . quote_plus ( c ) for c in self . categories ] ) <EOL> if len ( self . categories ) > <NUM_LIT:0> : <EOL> q_feed = q_feed + '<STR_LIT>' + category_string <EOL> return atom . service . BuildUri ( q_feed , self ) <EOL> def __str__ ( self ) : <EOL> return self . ToUri ( ) </s>
