diff --git a/aweber_api/__init__.py b/aweber_api/__init__.py index 4c38221..6db27bd 100644 --- a/aweber_api/__init__.py +++ b/aweber_api/__init__.py @@ -1,7 +1,13 @@ from urlparse import parse_qs -from aweber_api.base import (AWeberBase, API_BASE, ACCESS_TOKEN_URL, - REQUEST_TOKEN_URL, AUTHORIZE_URL, APIException) +from aweber_api.base import ( + ACCESS_TOKEN_URL, + APIException, + API_BASE, + AUTHORIZE_URL, + AWeberBase, + REQUEST_TOKEN_URL, +) from aweber_api.collection import AWeberCollection from aweber_api.entry import AWeberEntry from aweber_api.oauth import OAuthAdapter @@ -9,9 +15,13 @@ class AWeberAPI(AWeberBase): - """ Base class for connecting to the AWeberAPI. Created with a consumer key - and secret, then used to either generate tokens for authorizing a user, or - can be provided tokens and used to access that user's resources. """ + """Base class for connecting to the AWeberAPI. + + Created with a consumer key and secret, then used to either generate + tokens for authorizing a user, or can be provided tokens and used to + access that user's resources. + + """ def __init__(self, consumer_key, consumer_secret): self.adapter = OAuthAdapter(consumer_key, consumer_secret, API_BASE) @@ -19,25 +29,19 @@ def __init__(self, consumer_key, consumer_secret): @classmethod def parse_authorization_code(cls, authorization_code): - """ - Class method to exchange an authorization code for new api keys. - Returns a tuple containing the new consumer key/secret and access - token key/secret. + """Exchange an authorization code for new api keys. + + Returns a tuple containing the new consumer key/secret and + access token key/secret. + """ # parse and validate authorization code - keys = authorization_code.split('|') - if len(keys) < 5: - raise APIException('Invalid Authorization Code') - - # create an instance of AWeberAPI for getting the access token + keys = cls._parse_and_validate_authorization_code(authorization_code) consumer_key = keys[0] consumer_secret = keys[1] - instance = cls(consumer_key, consumer_secret) - # set request token and verifier code - instance.user.request_token = keys[2] - instance.user.token_secret = keys[3] - instance.user.verifier = keys[4] + # create an instance of AWeberAPI for getting the access token + instance = cls._create_new_instance(keys) # exchange request token for an access token access_key, access_secret = instance.get_access_token() @@ -45,78 +49,118 @@ def parse_authorization_code(cls, authorization_code): # return consumer key/secret and access token key/secret return consumer_key, consumer_secret, access_key, access_secret + @classmethod + def _parse_and_validate_authorization_code(cls, authorization_code): + """parse and validate authorization code.""" + keys = authorization_code.split('|') + if len(keys) < 5: + raise APIException('Invalid Authorization Code') + + return keys + + @classmethod + def _create_new_instance(cls, keys): + """Create an instance of AWeberAPI for getting the access token.""" + instance = cls(keys[0], keys[1]) + instance.user.request_token = keys[2] + instance.user.token_secret = keys[3] + instance.user.verifier = keys[4] + + return instance + @property def authorize_url(self): - """ - Returns the authorize url, potentially containing the request token - parameter + """Return the authorize url. + + Potentially containing the request token parameter. + """ if self.user.request_token: - return "{0}?oauth_token={1}".format(AUTHORIZE_URL, - self.user.request_token) + return "{0}?oauth_token={1}".format( + AUTHORIZE_URL, self.user.request_token) + return AUTHORIZE_URL def get_request_token(self, callback_url): + """Get a new request token / token secret for the callback url. + + Returns request token / secret, and sets properties on the + AWeberUser object (self.user). + """ - Gets a new request token / token secret for the given callback URL - and the current consumer. Returns token / secret, and sets properties - on the AWeberUser object (self.user) - """ - data = { 'oauth_callback' : callback_url } - response = self.adapter.request('POST', - REQUEST_TOKEN_URL, - data) - self.user.request_token, self.user.token_secret = self.\ - _parse_token_response(response) + data = {'oauth_callback': callback_url} + response = self.adapter.request( + 'POST', REQUEST_TOKEN_URL, data) + self.user.request_token, self.user.token_secret = ( + self._parse_token_response(response)) + return (self.user.request_token, self.user.token_secret) def get_access_token(self): + """Exchange request tokens for Access tokens. + + Gets an access token for the combination of + * request token + * token secret + * verifier + in the AWeberUser object at self.user. + + Updates the user object and returns the tokens. + """ - Gets an access token for the given request token / token secret / - verifier combination in the AWeberUser object at self.user - Updates the user object and returns the tokens - """ + data = {'oauth_verifier': self.user.verifier} + response = self.adapter.request( + 'POST', ACCESS_TOKEN_URL, data) + self.user.access_token, self.user.token_secret = ( + self._parse_token_response(response)) - data = { 'oauth_verifier' : self.user.verifier } - response = self.adapter.request('POST', - ACCESS_TOKEN_URL, - data) - self.user.access_token, self.user.token_secret = self.\ - _parse_token_response(response) return (self.user.access_token, self.user.token_secret) def _parse_token_response(self, response): - if not type(response) == str: + """Parses token response. + + Return the token key and the token secret + + """ + if not isinstance(response, str): raise TypeError('Expected response to be a string') data = parse_qs(response) - if not 'oauth_token' in data and not 'oauth_token_secret' in data: + if not data.get('oauth_token') or not data.get('oauth_token_secret'): raise ValueError('OAuth parameters not returned') + return (data['oauth_token'][0], data['oauth_token_secret'][0]) def get_account(self, access_token=False, token_secret=False): - """ - Returns the AWeberEntry object for the account specified by the - access_token and token_secret currently in the self.user object. - Optionally, access_token and token_secret can be provided to replace - the properties in self.user.access_token and self.user.token_secret, - respectively. + """Returns the AWeberEntry object for the account. + + Specified by the access_token and token_secret currently + in the self.user object. + + Optionally, access_token and token_secret can be provided to + replace the properties in self.user.access_token and + self.user.token_secret, respectively. + """ if access_token: self.user.access_token = access_token if token_secret: self.user.token_secret = token_secret + url = '/accounts' response = self.adapter.request('GET', url) accounts = self._read_response(url, response) + return accounts[0] class AWeberUser(object): - """ - Simple data storage object representing the user in the OAuth model. Has - properties for request_token, token_secret, access_token, and verifier. + """Data storage object representing the user in the OAuth model. + + Has properties for request_token, token_secret, access_token, and + verifier. + """ request_token = None @@ -125,4 +169,5 @@ class AWeberUser(object): verifier = None def get_highest_priority_token(self): + """Return either the access token or the request token.""" return self.access_token or self.request_token diff --git a/aweber_api/base.py b/aweber_api/base.py index 2bb2c84..0432563 100644 --- a/aweber_api/base.py +++ b/aweber_api/base.py @@ -1,8 +1,7 @@ -API_BASE = 'https://api.aweber.com/1.0' - ACCESS_TOKEN_URL = 'https://auth.aweber.com/1.0/oauth/access_token' -REQUEST_TOKEN_URL = 'https://auth.aweber.com/1.0/oauth/request_token' +API_BASE = 'https://api.aweber.com/1.0' AUTHORIZE_URL = 'https://auth.aweber.com/1.0/oauth/authorize' +REQUEST_TOKEN_URL = 'https://auth.aweber.com/1.0/oauth/request_token' class APIException(Exception): @@ -10,25 +9,28 @@ class APIException(Exception): class AWeberBase(object): - """ - Provides functionality shared accross all AWeber objects - """ + """Provides functionality shared accross all AWeber objects""" collections_map = { - 'account' : ['lists', 'integrations'], - 'broadcast_campaign' : ['links', 'messages', 'stats'], + 'account': ['lists', 'integrations'], + 'broadcast_campaign': ['links', 'messages', 'stats'], 'component': [], 'custom_field': [], - 'followup_campaign' : ['links', 'messages', 'stats'], + 'followup_campaign': ['links', 'messages', 'stats'], 'integration': [], - 'link' : ['clicks'], - 'list' : ['campaigns', 'custom_fields', 'subscribers', - 'web_forms', 'web_form_split_tests'], - 'message' : ['opens', 'tracked_events'], + 'link': ['clicks'], + 'list': [ + 'campaigns', + 'custom_fields', + 'subscribers', + 'web_forms', + 'web_form_split_tests', + ], + 'message': ['opens', 'tracked_events'], 'service-root': 'accounts', - 'subscriber' : [], + 'subscriber': [], 'tracked_events': [], 'web_form': [], - 'web_form_split_test' : ['components'] + 'web_form_split_test': ['components'], } @property @@ -36,9 +38,7 @@ def user(self): return self.adapter.user def load_from_url(self, url): - """ - Gets an AWeberCollection or AWeberEntry from the given URL. - """ + """Gets an AWeberCollection or AWeberEntry from a given URL.""" response = self.adapter.request('GET', url) return self._read_response(url, response) @@ -50,15 +50,22 @@ def _read_response(self, url, response): if 'entries' in response: from aweber_api.collection import AWeberCollection return AWeberCollection(url, response, self.adapter) + if 'resource_type_link' in response: from aweber_api.entry import AWeberEntry return AWeberEntry(url, response, self.adapter) + raise TypeError('Unknown value returned') def _parseNamedOperation(self, data): from aweber_api.entry import AWeberEntry entries = [] for item in data: - entries.append(AWeberEntry(item['self_link'].replace(API_BASE, ''), - item, self.adapter)) + entries.append( + AWeberEntry( + item['self_link'].replace(API_BASE, ''), + item, + self.adapter, + ) + ) return entries