diff --git a/.gitignore b/.gitignore index 60d1cc6..b9726bc 100644 --- a/.gitignore +++ b/.gitignore @@ -41,3 +41,4 @@ docs/_build test.py .venv +.idea \ No newline at end of file diff --git a/tests/test_core_ads.py b/tests/test_core_ads.py deleted file mode 100644 index 732b0eb..0000000 --- a/tests/test_core_ads.py +++ /dev/null @@ -1,23 +0,0 @@ -from .config import ( - test_tweet_object, test_tweet_html, unittest -) - -import responses -import requests -from twython.api_ads import TwythonAds - -from twython.compat import is_py2 -if is_py2: - from StringIO import StringIO -else: - from io import StringIO - -try: - import unittest.mock as mock -except ImportError: - import mock - - -class TwythonAPITestCase(unittest.TestCase): - def setUp(self): - self.api = TwythonAds('', '', '', '') \ No newline at end of file diff --git a/tests/test_endpoints_ads.py b/tests/test_endpoints_ads.py index 780b9eb..9a4543e 100644 --- a/tests/test_endpoints_ads.py +++ b/tests/test_endpoints_ads.py @@ -1,13 +1,11 @@ import base64 import datetime import urllib - from twython import Twython, TwythonError from .config import ( app_key, app_secret, oauth_token, oauth_token_secret, access_token, test_account_id, test_funding_instrument_id, test_campaign_id, unittest ) -from twython.api_ads import TwythonAds class TwythonEndpointsTestCase(unittest.TestCase): @@ -42,36 +40,36 @@ class TwythonEndpointsTestCase(unittest.TestCase): 'headers': {} } - self.api = TwythonAds(app_key, app_secret, - oauth_token, oauth_token_secret, - client_args=client_args) + self.api = Twython(app_key, app_secret, + oauth_token, oauth_token_secret, + client_args=client_args) self.oauth2_api = Twython(app_key, access_token=access_token, client_args=oauth2_client_args) - @unittest.skip('skipping non-updated test') + # @unittest.skip('skipping non-updated test') def test_get_accounts(self): accounts = self.api.get_accounts() self.assertTrue(len(accounts) >= 0) - @unittest.skip('skipping non-updated test') + # @unittest.skip('skipping non-updated test') def test_get_account(self): account = self.api.get_account(test_account_id) self.assertEqual(account['id'], test_account_id) with self.assertRaises(TwythonError): self.api.get_account('1234') - @unittest.skip('skipping non-updated test') + # @unittest.skip('skipping non-updated test') def test_get_account_features(self): account_features = self.api.get_account_features(test_account_id) self.assertTrue(len(account_features) >= 0) - @unittest.skip('skipping non-updated test') + # @unittest.skip('skipping non-updated test') def test_get_funding_instruments(self): funding_instruments = self.api.get_funding_instruments(test_account_id) self.assertTrue(len(funding_instruments) >= 0) - @unittest.skip('skipping non-updated test') + # @unittest.skip('skipping non-updated test') def test_get_funding_instrument(self): funding_instrument = self.api.get_funding_instrument(test_account_id, test_funding_instrument_id) self.assertEqual(funding_instrument['id'], test_funding_instrument_id) @@ -79,17 +77,17 @@ class TwythonEndpointsTestCase(unittest.TestCase): with self.assertRaises(TwythonError): self.api.get_funding_instrument('1234', '1234') - @unittest.skip('skipping non-updated test') + # @unittest.skip('skipping non-updated test') def test_get_iab_categories(self): iab_categories = self.api.get_iab_categories() self.assertTrue(len(iab_categories) >= 0) - @unittest.skip('skipping non-updated test') + # @unittest.skip('skipping non-updated test') def test_get_available_platforms(self): available_platforms = self.api.get_available_platforms() self.assertTrue(len(available_platforms) >= 0) - @unittest.skip('skipping non-updated test') + # @unittest.skip('skipping non-updated test') def test_get_available_locations(self): params = { 'location_type': 'CITY', @@ -98,12 +96,12 @@ class TwythonEndpointsTestCase(unittest.TestCase): available_locations = self.api.get_available_locations(**params) self.assertTrue(len(available_locations) > 0) - @unittest.skip('skipping non-updated test') + # @unittest.skip('skipping non-updated test') def test_get_campaigns(self): campaigns = self.api.get_campaigns(test_account_id) self.assertTrue(len(campaigns) >= 0) - @unittest.skip('skipping non-updated test') + # @unittest.skip('skipping non-updated test') def test_create_and_delete_campaign(self): campaign_id = self._create_test_campaign() campaign_check = self.api.get_campaign(test_account_id, campaign_id) @@ -121,7 +119,7 @@ class TwythonEndpointsTestCase(unittest.TestCase): is_deleted = self.api.delete_campaign(test_account_id, campaign_id) self.assertTrue(is_deleted) - @unittest.skip('skipping non-updated test') + # @unittest.skip('skipping non-updated test') def test_create_and_delete_line_item(self): campaign_id = self._create_test_campaign() line_item_id = self._create_test_line_item(campaign_id) @@ -142,7 +140,7 @@ class TwythonEndpointsTestCase(unittest.TestCase): is_deleted = self.api.delete_line_item(test_account_id, line_item_id) self.assertTrue(is_deleted) - @unittest.skip('skipping non-updated test') + # @unittest.skip('skipping non-updated test') def test_upload_image(self): response = self._upload_test_image() self.assertIsNotNone(response['media_id']) @@ -158,12 +156,12 @@ class TwythonEndpointsTestCase(unittest.TestCase): response = self.api.upload_image(**upload_data) return response - @unittest.skip('skipping non-updated test') + # @unittest.skip('skipping non-updated test') def test_get_website_cards(self): response = self.api.get_website_cards(test_account_id) self.assertTrue(len(response) >= 0) - @unittest.skip('skipping non-updated test') + # @unittest.skip('skipping non-updated test') def test_create_and_delete_website_card(self): card_id = self._create_test_website_card() card = self.api.get_website_card(test_account_id, card_id) @@ -189,7 +187,7 @@ class TwythonEndpointsTestCase(unittest.TestCase): response_delete = self.api.delete_website_card(test_account_id, card_id) self.assertEqual(response_delete['id'], card_id) - @unittest.skip('skipping non-updated test') + # @unittest.skip('skipping non-updated test') def test_create_promoted_only_tweet(self): card_id, tweet_id = self._create_test_promoted_only_tweet() self._delete_test_website_card(card_id) @@ -206,7 +204,7 @@ class TwythonEndpointsTestCase(unittest.TestCase): self.assertIsNotNone(tweet_id) return card_id, tweet_id - @unittest.skip('skipping non-updated test') + # @unittest.skip('skipping non-updated test') def test_promote_and_unpromote_tweet(self): campaign_id = self._create_test_campaign() line_item_id = self._create_test_line_item(campaign_id) @@ -228,7 +226,7 @@ class TwythonEndpointsTestCase(unittest.TestCase): self._delete_test_campaign(campaign_id) self._delete_test_website_card(card_id) - @unittest.skip('skipping non-updated test') + # @unittest.skip('skipping non-updated test') def test_add_targeting_criteria(self): campaign_id = self._create_test_campaign() line_item_id = self._create_test_line_item(campaign_id) @@ -254,7 +252,7 @@ class TwythonEndpointsTestCase(unittest.TestCase): self.assertEquals(response_add['line_item_id'], line_item_id) return response_add['id'] - @unittest.skip('skipping non-updated test') + # @unittest.skip('skipping non-updated test') def test_get_stats_promoted_tweets(self): line_items = self.api.get_line_items(test_account_id, test_campaign_id) promoted_tweets = self.api.get_promoted_tweets(test_account_id, line_items[0]['id']) diff --git a/twython/api.py b/twython/api.py index 2ed50bc..add9267 100644 --- a/twython/api.py +++ b/twython/api.py @@ -5,8 +5,8 @@ twython.api ~~~~~~~~~~~ This module contains functionality for access to core Twitter API calls, -Twitter Authentication, and miscellaneous methods that are useful when -dealing with the Twitter API +Twitter Ads API calls, Twitter Authentication, and miscellaneous methods +that are useful when dealing with the Twitter API """ import warnings @@ -20,17 +20,19 @@ from . import __version__ from .advisory import TwythonDeprecationWarning from .compat import json, urlencode, parse_qsl, quote_plus, str, is_py2 from .endpoints import EndpointsMixin +from .endpoints_ads import EndpointsAdsMixin from .exceptions import TwythonError, TwythonAuthError, TwythonRateLimitError from .helpers import _transparent_params +from .api_type import API_TYPE_TWITTER, API_TYPE_TWITTER_ADS warnings.simplefilter('always', TwythonDeprecationWarning) # For Python 2.7 > -class Twython(EndpointsMixin, object): +class Twython(EndpointsMixin, EndpointsAdsMixin, object): def __init__(self, app_key=None, app_secret=None, oauth_token=None, oauth_token_secret=None, access_token=None, token_type='bearer', oauth_version=1, api_version='1.1', - client_args=None, auth_endpoint='authenticate'): + api_ads_version='0', client_args=None, auth_endpoint='authenticate'): """Instantiates an instance of Twython. Takes optional parameters for authentication and such (see below). @@ -48,6 +50,8 @@ class Twython(EndpointsMixin, object): Default: 1 :param api_version: (optional) Choose which Twitter API version to use. Default: 1.1 + :param api_ads_version: (optional) Choose which Twitter Ads API version to + use. Default: 0 :param client_args: (optional) Accepts some requests Session parameters and some requests Request parameters. @@ -64,7 +68,9 @@ class Twython(EndpointsMixin, object): # API urls, OAuth urls and API version; needed for hitting that there # API. self.api_version = api_version + self.api_ads_version = api_ads_version self.api_url = 'https://api.twitter.com/%s' + self.api_ads_url = 'https://ads-api.twitter.com/%s' self.app_key = app_key self.app_secret = app_secret @@ -224,7 +230,7 @@ class Twython(EndpointsMixin, object): return error_message - def request(self, endpoint, method='GET', params=None, version='1.1'): + def request(self, endpoint, api_type=API_TYPE_TWITTER, method='GET', params=None, version='1.1'): """Return dict of response received from Twitter's API :param endpoint: (required) Full url or Twitter API endpoint @@ -251,6 +257,8 @@ class Twython(EndpointsMixin, object): # i.e. https://api.twitter.com/1.1/search/tweets.json if endpoint.startswith('https://'): url = endpoint + elif api_type == API_TYPE_TWITTER_ADS: + url = '%s/%s' % (self.api_ads_url % version, endpoint) else: url = '%s/%s.json' % (self.api_url % version, endpoint) @@ -259,13 +267,17 @@ class Twython(EndpointsMixin, object): return content - def get(self, endpoint, params=None, version='1.1'): + def get(self, endpoint, api_type=API_TYPE_TWITTER, params=None, version='1.1'): """Shortcut for GET requests via :class:`request`""" - return self.request(endpoint, params=params, version=version) + return self.request(endpoint, api_type=api_type, params=params, version=version) - def post(self, endpoint, params=None, version='1.1'): + def post(self, endpoint, api_type=API_TYPE_TWITTER, params=None, version='1.1'): """Shortcut for POST requests via :class:`request`""" - return self.request(endpoint, 'POST', params=params, version=version) + return self.request(endpoint, api_type=api_type, method='POST', params=params, version=version) + + def delete(self, endpoint, api_type=API_TYPE_TWITTER, params=None, version='1.1'): + """Shortcut for DELETE requests via :class:`request`""" + return self.request(endpoint, api_type=api_type, method='DELETE', params=params, version=version) def get_lastfunction_header(self, header, default_return_value=None): """Returns a specific header from the last API call diff --git a/twython/api_ads.py b/twython/api_ads.py deleted file mode 100644 index e0e5b8f..0000000 --- a/twython/api_ads.py +++ /dev/null @@ -1,461 +0,0 @@ -# -*- coding: utf-8 -*- - -""" -twython.api_ads -~~~~~~~~~~~ - -This module contains functionality for access to core Twitter Ads API calls, -Twitter Authentication, and miscellaneous methods that are useful when -dealing with the Twitter Ads API -""" - -import warnings -import re - -import requests -from requests.auth import HTTPBasicAuth -from requests_oauthlib import OAuth1, OAuth2 - -from . import __version__ -from .advisory import TwythonDeprecationWarning -from .compat import json, urlencode, parse_qsl, quote_plus, str, is_py2 -from .exceptions import TwythonError, TwythonAuthError, TwythonRateLimitError -from .helpers import _transparent_params -from twython.endpoints_ads import EndpointsAdsMixin - -warnings.simplefilter('always', TwythonDeprecationWarning) # For Python 2.7 > - - -class TwythonAds(EndpointsAdsMixin, object): - def __init__(self, app_key=None, app_secret=None, oauth_token=None, - oauth_token_secret=None, access_token=None, - token_type='bearer', oauth_version=1, api_version='0', - client_args=None, auth_endpoint='authenticate'): - """Instantiates an instance of TwythonAds. Takes optional parameters for - authentication and such (see below). - - :param app_key: (optional) Your applications key - :param app_secret: (optional) Your applications secret key - :param oauth_token: (optional) When using **OAuth 1**, combined with - oauth_token_secret to make authenticated calls - :param oauth_token_secret: (optional) When using **OAuth 1** combined - with oauth_token to make authenticated calls - :param access_token: (optional) When using **OAuth 2**, provide a - valid access token if you have one - :param token_type: (optional) When using **OAuth 2**, provide your - token type. Default: bearer - :param oauth_version: (optional) Choose which OAuth version to use. - Default: 1 - :param api_version: (optional) Choose which Twitter API version to - use. Default: 0 - - :param client_args: (optional) Accepts some requests Session parameters - and some requests Request parameters. - See http://docs.python-requests.org/en/latest/api/#sessionapi - and requests section below it for details. - [ex. headers, proxies, verify(SSL verification)] - :param auth_endpoint: (optional) Lets you select which authentication - endpoint will use your application. - This will allow the application to have DM access - if the endpoint is 'authorize'. - Default: authenticate. - """ - - # API urls, OAuth urls and API version; needed for hitting that there - # API. - self.api_version = api_version - self.api_url = 'https://ads-api.twitter.com/%s' - - self.app_key = app_key - self.app_secret = app_secret - self.oauth_token = oauth_token - self.oauth_token_secret = oauth_token_secret - self.access_token = access_token - - # OAuth 1 - self.request_token_url = self.api_url % 'oauth/request_token' - self.access_token_url = self.api_url % 'oauth/access_token' - self.authenticate_url = self.api_url % ('oauth/%s' % auth_endpoint) - - if self.access_token: # If they pass an access token, force OAuth 2 - oauth_version = 2 - - self.oauth_version = oauth_version - - # OAuth 2 - if oauth_version == 2: - self.request_token_url = self.api_url % 'oauth2/token' - - self.client_args = client_args or {} - default_headers = {'User-Agent': 'Twython v' + __version__} - if 'headers' not in self.client_args: - # If they didn't set any headers, set our defaults for them - self.client_args['headers'] = default_headers - elif 'User-Agent' not in self.client_args['headers']: - # If they set headers, but didn't include User-Agent.. set - # it for them - self.client_args['headers'].update(default_headers) - - # Generate OAuth authentication object for the request - # If no keys/tokens are passed to __init__, auth=None allows for - # unauthenticated requests, although I think all v1.1 requests - # need auth - auth = None - if oauth_version == 1: - # User Authentication is through OAuth 1 - if self.app_key is not None and self.app_secret is not None: - auth = OAuth1(self.app_key, self.app_secret, - self.oauth_token, self.oauth_token_secret) - - elif oauth_version == 2 and self.access_token: - # Application Authentication is through OAuth 2 - token = {'token_type': token_type, - 'access_token': self.access_token} - auth = OAuth2(self.app_key, token=token) - - self.client = requests.Session() - self.client.auth = auth - - # Make a copy of the client args and iterate over them - # Pop out all the acceptable args at this point because they will - # Never be used again. - client_args_copy = self.client_args.copy() - for k, v in client_args_copy.items(): - if k in ('cert', 'hooks', 'max_redirects', 'proxies'): - setattr(self.client, k, v) - self.client_args.pop(k) # Pop, pop! - - # Headers are always present, so we unconditionally pop them and merge - # them into the session headers. - self.client.headers.update(self.client_args.pop('headers')) - - self._last_call = None - - def __repr__(self): - return '' % (self.app_key) - - def _request(self, url, method='GET', params=None, api_call=None): - """Internal request method""" - method = method.lower() - params = params or {} - - func = getattr(self.client, method) - params, files = _transparent_params(params) - - requests_args = {} - for k, v in self.client_args.items(): - # Maybe this should be set as a class variable and only done once? - if k in ('timeout', 'allow_redirects', 'stream', 'verify'): - requests_args[k] = v - - if method == 'get': - requests_args['params'] = params - else: - requests_args.update({ - 'data': params, - 'files': files, - }) - try: - response = func(url, **requests_args) - except requests.RequestException as e: - raise TwythonError(str(e)) - - # create stash for last function intel - self._last_call = { - 'api_call': api_call, - 'api_error': None, - 'cookies': response.cookies, - 'headers': response.headers, - 'status_code': response.status_code, - 'url': response.url, - 'content': response.text, - } - - # greater than 304 (not modified) is an error - if response.status_code > 304: - error_message = self._get_error_message(response) - self._last_call['api_error'] = error_message - - ExceptionType = TwythonError - if response.status_code == 429: - # Twitter API 1.1, always return 429 when - # rate limit is exceeded - ExceptionType = TwythonRateLimitError - elif response.status_code == 401 or 'Bad Authentication data' \ - in error_message: - # Twitter API 1.1, returns a 401 Unauthorized or - # a 400 "Bad Authentication data" for invalid/expired - # app keys/user tokens - ExceptionType = TwythonAuthError - - raise ExceptionType( - error_message, - error_code=response.status_code, - retry_after=response.headers.get('X-Rate-Limit-Reset')) - - try: - if response.status_code == 204: - content = response.content - else: - content = response.json() - except ValueError: - raise TwythonError('Response was not valid JSON. \ - Unable to decode.') - - return content - - def _get_error_message(self, response): - """Parse and return the first error message""" - - error_message = 'An error occurred processing your request.' - try: - content = response.json() - # {"errors":[{"code":34,"message":"Sorry, - # that page does not exist"}]} - error_message = content['errors'][0]['message'] - except TypeError: - error_message = content['errors'] - except ValueError: - # bad json data from Twitter for an error - pass - except (KeyError, IndexError): - # missing data so fallback to default message - pass - - return error_message - - def request(self, endpoint, method='GET', params=None, version='0'): - """Return dict of response received from Twitter's API - - :param endpoint: (required) Full url or Twitter API endpoint - (e.g. search/tweets) - :type endpoint: string - :param method: (optional) Method of accessing data, either - GET or POST. (default GET) - :type method: string - :param params: (optional) Dict of parameters (if any) accepted - the by Twitter API endpoint you are trying to - access (default None) - :type params: dict or None - :param version: (optional) Twitter API version to access - (default 0) - :type version: string - - :rtype: dict - """ - - if endpoint.startswith('http://'): - raise TwythonError('ads-api.twitter.com is restricted to SSL/TLS traffic.') - - # In case they want to pass a full Twitter URL - # i.e. https://api.twitter.com/1.1/search/tweets.json - if endpoint.startswith('https://'): - url = endpoint - else: - url = '%s/%s' % (self.api_url % version, endpoint) - - content = self._request(url, method=method, params=params, - api_call=url) - - return content - - def get(self, endpoint, params=None, version='0'): - """Shortcut for GET requests via :class:`request`""" - return self.request(endpoint, params=params, version=version) - - def post(self, endpoint, params=None, version='0'): - """Shortcut for POST requests via :class:`request`""" - return self.request(endpoint, 'POST', params=params, version=version) - - def delete(self, endpoint, params=None, version='0'): - """Shortcut for DELETE requests via :class:`request`""" - return self.request(endpoint, 'DELETE', params=params, version=version) - - def get_lastfunction_header(self, header, default_return_value=None): - """Returns a specific header from the last API call - This will return None if the header is not present - - :param header: (required) The name of the header you want to get - the value of - - Most useful for the following header information: - x-rate-limit-limit, - x-rate-limit-remaining, - x-rate-limit-class, - x-rate-limit-reset - - """ - if self._last_call is None: - raise TwythonError('This function must be called after an API call. \ - It delivers header information.') - - return self._last_call['headers'].get(header, default_return_value) - - def get_authentication_tokens(self, callback_url=None, force_login=False, - screen_name=''): - """Returns a dict including an authorization URL, ``auth_url``, to - direct a user to - - :param callback_url: (optional) Url the user is returned to after - they authorize your app (web clients only) - :param force_login: (optional) Forces the user to enter their - credentials to ensure the correct users - account is authorized. - :param screen_name: (optional) If forced_login is set OR user is - not currently logged in, Prefills the username - input box of the OAuth login screen with the - given value - - :rtype: dict - """ - if self.oauth_version != 1: - raise TwythonError('This method can only be called when your \ - OAuth version is 1.0.') - - request_args = {} - if callback_url: - request_args['oauth_callback'] = callback_url - response = self.client.get(self.request_token_url, params=request_args) - - if response.status_code == 401: - raise TwythonAuthError(response.content, - error_code=response.status_code) - elif response.status_code != 200: - raise TwythonError(response.content, - error_code=response.status_code) - - request_tokens = dict(parse_qsl(response.content.decode('utf-8'))) - if not request_tokens: - raise TwythonError('Unable to decode request tokens.') - - oauth_callback_confirmed = request_tokens.get('oauth_callback_confirmed') \ - == 'true' - - auth_url_params = { - 'oauth_token': request_tokens['oauth_token'], - } - - if force_login: - auth_url_params.update({ - 'force_login': force_login, - 'screen_name': screen_name - }) - - # Use old-style callback argument if server didn't accept new-style - if callback_url and not oauth_callback_confirmed: - auth_url_params['oauth_callback'] = self.callback_url - - request_tokens['auth_url'] = self.authenticate_url + \ - '?' + urlencode(auth_url_params) - - return request_tokens - - def get_authorized_tokens(self, oauth_verifier): - """Returns a dict of authorized tokens after they go through the - :class:`get_authentication_tokens` phase. - - :param oauth_verifier: (required) The oauth_verifier (or a.k.a PIN - for non web apps) retrieved from the callback url querystring - :rtype: dict - - """ - if self.oauth_version != 1: - raise TwythonError('This method can only be called when your \ - OAuth version is 1.0.') - - response = self.client.get(self.access_token_url, - params={'oauth_verifier': oauth_verifier}, - headers={'Content-Type': 'application/\ - json'}) - - if response.status_code == 401: - try: - try: - # try to get json - content = response.json() - except AttributeError: # pragma: no cover - # if unicode detected - content = json.loads(response.content) - except ValueError: - content = {} - - raise TwythonError(content.get('error', 'Invalid / expired To \ - ken'), error_code=response.status_code) - - authorized_tokens = dict(parse_qsl(response.content.decode('utf-8'))) - if not authorized_tokens: - raise TwythonError('Unable to decode authorized tokens.') - - return authorized_tokens # pragma: no cover - - def obtain_access_token(self): - """Returns an OAuth 2 access token to make OAuth 2 authenticated - read-only calls. - - :rtype: string - """ - if self.oauth_version != 2: - raise TwythonError('This method can only be called when your \ - OAuth version is 2.0.') - - data = {'grant_type': 'client_credentials'} - basic_auth = HTTPBasicAuth(self.app_key, self.app_secret) - try: - response = self.client.post(self.request_token_url, - data=data, auth=basic_auth) - content = response.content.decode('utf-8') - try: - content = content.json() - except AttributeError: - content = json.loads(content) - access_token = content['access_token'] - except (KeyError, ValueError, requests.exceptions.RequestException): - raise TwythonAuthError('Unable to obtain OAuth 2 access token.') - else: - return access_token - - @staticmethod - def construct_api_url(api_url, **params): - """Construct a Twitter API url, encoded, with parameters - - :param api_url: URL of the Twitter API endpoint you are attempting - to construct - :param \*\*params: Parameters that are accepted by Twitter for the - endpoint you're requesting - :rtype: string - - Usage:: - - >>> from twython import Twython - >>> twitter = Twython() - - >>> api_url = 'https://api.twitter.com/1.1/search/tweets.json' - >>> constructed_url = twitter.construct_api_url(api_url, q='python', - result_type='popular') - >>> print constructed_url - https://api.twitter.com/1.1/search/tweets.json?q=python&result_type=popular - - """ - querystring = [] - params, _ = _transparent_params(params or {}) - params = requests.utils.to_key_val_list(params) - for (k, v) in params: - querystring.append( - '%s=%s' % (TwythonAds.encode(k), quote_plus(TwythonAds.encode(v))) - ) - return '%s?%s' % (api_url, '&'.join(querystring)) - - @staticmethod - def unicode2utf8(text): - try: - if is_py2 and isinstance(text, str): - text = text.encode('utf-8') - except: - pass - return text - - @staticmethod - def encode(text): - if is_py2 and isinstance(text, (str)): - return TwythonAds.unicode2utf8(text) - return str(text) diff --git a/twython/api_type.py b/twython/api_type.py new file mode 100644 index 0000000..9778603 --- /dev/null +++ b/twython/api_type.py @@ -0,0 +1,2 @@ +API_TYPE_TWITTER='api' +API_TYPE_TWITTER_ADS='api_ads' diff --git a/twython/endpoints_ads.py b/twython/endpoints_ads.py index c99dfd8..9fe1f3c 100644 --- a/twython/endpoints_ads.py +++ b/twython/endpoints_ads.py @@ -15,6 +15,8 @@ The API functions that are implemented in this module are documented at: https://dev.twitter.com/ads/overview """ +from .api_type import API_TYPE_TWITTER_ADS + try: from StringIO import StringIO except ImportError: @@ -23,118 +25,143 @@ except ImportError: class EndpointsAdsMixin(object): def get_accounts(self, **params): - response = self.get('accounts', params=params) + response = self.get('accounts', params=params, api_type=API_TYPE_TWITTER_ADS, version=self.api_ads_version) return response['data'] def get_account(self, account_id, **params): - response = self.get('accounts/%s' % account_id, params=params) + response = self.get('accounts/%s' % account_id, params=params, api_type=API_TYPE_TWITTER_ADS, + version=self.api_ads_version) return response['data'] def get_account_features(self, account_id, **params): - response = self.get('accounts/%s/features' % account_id, params=params) + response = self.get('accounts/%s/features' % account_id, params=params, api_type=API_TYPE_TWITTER_ADS, + version=self.api_ads_version) return response['data'] def get_funding_instruments(self, account_id, **params): - response = self.get('accounts/%s/funding_instruments' % account_id, params=params) + response = self.get('accounts/%s/funding_instruments' % account_id, params=params, + api_type=API_TYPE_TWITTER_ADS, version=self.api_ads_version) return response['data'] def get_funding_instrument(self, account_id, funding_instrument_id, **params): - response = self.get('accounts/%s/funding_instruments/%s' % (account_id, funding_instrument_id), params=params) + response = self.get('accounts/%s/funding_instruments/%s' % (account_id, funding_instrument_id), params=params, + api_type=API_TYPE_TWITTER_ADS, version=self.api_ads_version) return response['data'] def get_iab_categories(self, **params): - response = self.get('iab_categories', params=params) + response = self.get('iab_categories', params=params, api_type=API_TYPE_TWITTER_ADS, + version=self.api_ads_version) return response['data'] def get_available_platforms(self, **params): - response = self.get('targeting_criteria/platforms', params=params) + response = self.get('targeting_criteria/platforms', params=params, api_type=API_TYPE_TWITTER_ADS, + version=self.api_ads_version) return response['data'] def get_available_locations(self, **params): - response = self.get('targeting_criteria/locations', params=params) + response = self.get('targeting_criteria/locations', params=params, api_type=API_TYPE_TWITTER_ADS, + version=self.api_ads_version) return response['data'] def get_campaigns(self, account_id, **params): - response = self.get('accounts/%s/campaigns' % account_id, params=params) + response = self.get('accounts/%s/campaigns' % account_id, params=params, api_type=API_TYPE_TWITTER_ADS, + version=self.api_ads_version) return response['data'] def get_campaign(self, account_id, campaign_id, **params): - response = self.get('accounts/%s/campaigns/%s' % (account_id, campaign_id), params=params) + response = self.get('accounts/%s/campaigns/%s' % (account_id, campaign_id), params=params, + api_type=API_TYPE_TWITTER_ADS, version=self.api_ads_version) return response['data'] def create_campaign(self, account_id, **params): - response = self.post('accounts/%s/campaigns' % account_id, params=params) + response = self.post('accounts/%s/campaigns' % account_id, params=params, api_type=API_TYPE_TWITTER_ADS, + version=self.api_ads_version) return response['data'] def delete_campaign(self, account_id, campaign_id): - response = self.delete('accounts/%s/campaigns/%s' % (account_id, campaign_id)) + response = self.delete('accounts/%s/campaigns/%s' % (account_id, campaign_id), api_type=API_TYPE_TWITTER_ADS, + version=self.api_ads_version) return response['data']['deleted'] def create_line_item(self, account_id, campaign_id, **params): params_extended = params.copy() params_extended['campaign_id'] = campaign_id - response = self.post('accounts/%s/line_items' % account_id, params=params_extended) + response = self.post('accounts/%s/line_items' % account_id, params=params_extended, + api_type=API_TYPE_TWITTER_ADS, version=self.api_ads_version) return response['data'] def delete_line_item(self, account_id, line_item_id): - response = self.delete('accounts/%s/line_items/%s' % (account_id, line_item_id)) + response = self.delete('accounts/%s/line_items/%s' % (account_id, line_item_id), api_type=API_TYPE_TWITTER_ADS, + version=self.api_ads_version) return response['data']['deleted'] def get_line_items(self, account_id, campaign_id=None, **params): params_extended = params.copy() if campaign_id is not None: params_extended['campaign_ids'] = campaign_id - response = self.get('accounts/%s/line_items' % account_id, params=params_extended) + response = self.get('accounts/%s/line_items' % account_id, params=params_extended, + api_type=API_TYPE_TWITTER_ADS, version=self.api_ads_version) return response['data'] def get_website_cards(self, account_id, **params): - response = self.get('accounts/%s/cards/website' % account_id, params=params) + response = self.get('accounts/%s/cards/website' % account_id, params=params, api_type=API_TYPE_TWITTER_ADS, + version=self.api_ads_version) return response['data'] def get_website_card(self, account_id, card_id, **params): - response = self.get('accounts/%s/cards/website/%s' % (account_id, card_id), params=params) + response = self.get('accounts/%s/cards/website/%s' % (account_id, card_id), params=params, + api_type=API_TYPE_TWITTER_ADS, version=self.api_ads_version) return response['data'] def create_website_card(self, account_id, **params): # TODO: handle the case where name, website_title, website_url are too long! - response = self.post('accounts/%s/cards/website' % account_id, params=params) + response = self.post('accounts/%s/cards/website' % account_id, params=params, api_type=API_TYPE_TWITTER_ADS, + version=self.api_ads_version) return response['data'] def delete_website_card(self, account_id, card_id, **params): - response = self.delete('accounts/%s/cards/website/%s' % (account_id, card_id), params=params) + response = self.delete('accounts/%s/cards/website/%s' % (account_id, card_id), params=params, + api_type=API_TYPE_TWITTER_ADS, version=self.api_ads_version) return response['data'] def upload_image(self, **params): - response = self.post('https://upload.twitter.com/1.1/media/upload.json', params=params) + response = self.post('https://upload.twitter.com/1.1/media/upload.json', params=params, + api_type=API_TYPE_TWITTER_ADS, version=self.api_ads_version) return response def create_promoted_only_tweet(self, account_id, **params): - response = self.post('accounts/%s/tweet' % account_id, params=params) + response = self.post('accounts/%s/tweet' % account_id, params=params, api_type=API_TYPE_TWITTER_ADS, + version=self.api_ads_version) return response['data'] def promote_tweet(self, account_id, **params): - response = self.post('accounts/%s/promoted_tweets' % account_id, params=params) + response = self.post('accounts/%s/promoted_tweets' % account_id, params=params, api_type=API_TYPE_TWITTER_ADS, + version=self.api_ads_version) return response['data'] def unpromote_tweet(self, account_id, promotion_id): - response = self.delete('accounts/%s/promoted_tweets/%s' % (account_id, promotion_id)) + response = self.delete('accounts/%s/promoted_tweets/%s' % (account_id, promotion_id), + api_type=API_TYPE_TWITTER_ADS, version=self.api_ads_version) return response['data'] def get_promoted_tweets(self, account_id, line_item_id=None, **params): params_extended = params.copy() if line_item_id is not None: params_extended['line_item_id'] = line_item_id - response = self.get('accounts/%s/promoted_tweets' % account_id, params=params_extended) + response = self.get('accounts/%s/promoted_tweets' % account_id, params=params_extended, + api_type=API_TYPE_TWITTER_ADS, version=self.api_ads_version) return response['data'] def add_targeting_criteria(self, account_id, line_item_id, **params): params_extended = params.copy() params_extended['line_item_id'] = line_item_id - response = self.post('accounts/%s/targeting_criteria' % account_id, params=params_extended) + response = self.post('accounts/%s/targeting_criteria' % account_id, params=params_extended, + api_type=API_TYPE_TWITTER_ADS, version=self.api_ads_version) return response['data'] def remove_targeting_criteria(self, account_id, criteria_id): - response = self.delete('accounts/%s/targeting_criteria/%s' % (account_id, criteria_id)) + response = self.delete('accounts/%s/targeting_criteria/%s' % (account_id, criteria_id), + api_type=API_TYPE_TWITTER_ADS, version=self.api_ads_version) return response['data'] def get_stats_promoted_tweets(self, account_id, promoted_tweet_ids, **params): @@ -146,6 +173,7 @@ class EndpointsAdsMixin(object): chunk = promoted_tweet_ids[i:i + max_chunk_size] params_extended = params.copy() params_extended['promoted_tweet_ids'] = ",".join(chunk) - response = self.get('stats/accounts/%s/promoted_tweets' % account_id, params=params_extended) + response = self.get('stats/accounts/%s/promoted_tweets' % account_id, params=params_extended, + api_type=API_TYPE_TWITTER_ADS, version=self.api_ads_version) stats.extend(response['data']) return stats