diff --git a/tests/config.py b/tests/config.py index 21baa69..8616085 100644 --- a/tests/config.py +++ b/tests/config.py @@ -28,3 +28,6 @@ test_list_owner_screen_name = os.environ.get('TEST_LIST_OWNER_SCREEN_NAME', test_tweet_object = {u'contributors': None, u'truncated': False, u'text': u'http://t.co/FCmXyI6VHd is a #cool site, lol! @mikehelmick shd #checkitout. Love, @__twython__ https://t.co/67pwRvY6z9 http://t.co/N6InAO4B71', u'in_reply_to_status_id': None, u'id': 349683012054683648, u'favorite_count': 0, u'source': u'web', u'retweeted': False, u'coordinates': None, u'entities': {u'symbols': [], u'user_mentions': [{u'id': 29251354, u'indices': [45, 57], u'id_str': u'29251354', u'screen_name': u'mikehelmick', u'name': u'Mike Helmick'}, {u'id': 1431865928, u'indices': [81, 93], u'id_str': u'1431865928', u'screen_name': u'__twython__', u'name': u'Twython'}], u'hashtags': [{u'indices': [28, 33], u'text': u'cool'}, {u'indices': [62, 73], u'text': u'checkitout'}], u'urls': [{u'url': u'http://t.co/FCmXyI6VHd', u'indices': [0, 22], u'expanded_url': u'http://google.com', u'display_url': u'google.com'}, {u'url': u'https://t.co/67pwRvY6z9', u'indices': [94, 117], u'expanded_url': u'https://github.com', u'display_url': u'github.com'}], u'media': [{u'id': 537884378513162240, u'id_str': u'537884378513162240', u'indices': [118, 140], u'media_url': u'http://pbs.twimg.com/media/B3by_g-CQAAhrO5.jpg', u'media_url_https': u'https://pbs.twimg.com/media/B3by_g-CQAAhrO5.jpg', u'url': u'http://t.co/N6InAO4B71', u'display_url': u'pic.twitter.com/N6InAO4B71', u'expanded_url': u'http://twitter.com/pingofglitch/status/537884380060844032/photo/1', u'type': u'photo', u'sizes': {u'large': {u'w': 1024, u'h': 640, u'resize': u'fit'}, u'thumb': {u'w': 150, u'h': 150, u'resize': u'crop'}, u'medium': {u'w': 600, u'h': 375, u'resize': u'fit'}, u'small': {u'w': 340, u'h': 212, u'resize': u'fit'}}}]}, u'in_reply_to_screen_name': None, u'id_str': u'349683012054683648', u'retweet_count': 0, u'in_reply_to_user_id': None, u'favorited': False, u'user': {u'follow_request_sent': False, u'profile_use_background_image': True, u'default_profile_image': True, u'id': 1431865928, u'verified': False, u'profile_text_color': u'333333', u'profile_image_url_https': u'https://si0.twimg.com/sticky/default_profile_images/default_profile_3_normal.png', u'profile_sidebar_fill_color': u'DDEEF6', u'entities': {u'description': {u'urls': []}}, u'followers_count': 1, u'profile_sidebar_border_color': u'C0DEED', u'id_str': u'1431865928', u'profile_background_color': u'3D3D3D', u'listed_count': 0, u'profile_background_image_url_https': u'https://si0.twimg.com/images/themes/theme1/bg.png', u'utc_offset': None, u'statuses_count': 2, u'description': u'', u'friends_count': 1, u'location': u'', u'profile_link_color': u'0084B4', u'profile_image_url': u'http://a0.twimg.com/sticky/default_profile_images/default_profile_3_normal.png', u'following': False, u'geo_enabled': False, u'profile_background_image_url': u'http://a0.twimg.com/images/themes/theme1/bg.png', u'screen_name': u'__twython__', u'lang': u'en', u'profile_background_tile': False, u'favourites_count': 0, u'name': u'Twython', u'notifications': False, u'url': None, u'created_at': u'Thu May 16 01:11:09 +0000 2013', u'contributors_enabled': False, u'time_zone': None, u'protected': False, u'default_profile': False, u'is_translator': False}, u'geo': None, u'in_reply_to_user_id_str': None, u'possibly_sensitive': False, u'lang': u'en', u'created_at': u'Wed Jun 26 00:18:21 +0000 2013', u'in_reply_to_status_id_str': None, u'place': None} test_tweet_html = 'google.com is a #cool site, lol! @mikehelmick shd #checkitout. Love, @__twython__ github.com pic.twitter.com/N6InAO4B71' + +test_account_id = os.environ.get('TEST_ACCOUNT_ID') +test_funding_instrument_id = os.environ.get('TEST_FUNDING_INSTRUMENT_ID') diff --git a/tests/test_core_ads.py b/tests/test_core_ads.py new file mode 100644 index 0000000..732b0eb --- /dev/null +++ b/tests/test_core_ads.py @@ -0,0 +1,23 @@ +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 new file mode 100644 index 0000000..3fa68ba --- /dev/null +++ b/tests/test_endpoints_ads.py @@ -0,0 +1,72 @@ +import datetime +from twython import Twython, TwythonError, TwythonAuthError + +from .config import ( + app_key, app_secret, oauth_token, oauth_token_secret, + protected_twitter_1, protected_twitter_2, screen_name, + test_tweet_id, test_list_slug, test_list_owner_screen_name, + access_token, test_tweet_object, test_tweet_html, test_account_id, test_funding_instrument_id, unittest +) + +import time +from twython.api_ads import TwythonAds + + +class TwythonEndpointsTestCase(unittest.TestCase): + def setUp(self): + + client_args = { + 'headers': { + 'User-Agent': '__twython__ Test' + }, + 'allow_redirects': False + } + + # This is so we can hit coverage that Twython sets + # User-Agent for us if none is supplied + oauth2_client_args = { + 'headers': {} + } + + self.api = TwythonAds(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) + + def test_get_accounts(self): + accounts = self.api.get_accounts() + self.assertTrue(len(accounts) > 0) + + 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') + + def test_get_funding_instruments(self): + funding_instruments = self.api.get_funding_instruments(test_account_id) + self.assertTrue(len(funding_instruments) > 0) + + 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) + self.assertEqual(funding_instrument['account_id'], test_account_id) + with self.assertRaises(TwythonError): + self.api.get_funding_instrument('1234', '1234') + + def test_get_campaigns(self): + campaigns = self.api.get_campaigns(test_account_id) + self.assertTrue(len(campaigns) > 0) + + def test_create_campaign(self): + new_campaign = { + 'name': 'Test Twitter campaign - Twython', + 'funding_instrument_id': test_funding_instrument_id, + 'start_time': datetime.datetime.now().strftime('%Y-%m-%dT%H:%M:%SZ'), + 'daily_budget_amount_local_micro': 10 * 1000000 + } + campaign = self.api.create_campaign(test_account_id, **new_campaign) + self.assertEqual(campaign['account_id'], test_account_id) + self.assertIsNotNone(campaign['id']) diff --git a/twython/api_ads.py b/twython/api_ads.py new file mode 100644 index 0000000..00794f9 --- /dev/null +++ b/twython/api_ads.py @@ -0,0 +1,457 @@ +# -*- 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 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/endpoints_ads.py b/twython/endpoints_ads.py new file mode 100644 index 0000000..65e3ee1 --- /dev/null +++ b/twython/endpoints_ads.py @@ -0,0 +1,49 @@ +# -*- coding: utf-8 -*- + +""" +twython.endpoints_ads +~~~~~~~~~~~~~~~~~ + +This module adds Twitter Ads API support to the Twython library. +This module provides a mixin for a :class:`TwythonAds ` instance. +Parameters that need to be embedded in the API url just need to be passed +as a keyword argument. + +e.g. TwythonAds.retweet(id=12345) + +The API functions that are implemented in this module are documented at: +https://dev.twitter.com/ads/overview +""" + +import os +import warnings +try: + from StringIO import StringIO +except ImportError: + from io import StringIO + + +class EndpointsAdsMixin(object): + def get_accounts(self, **params): + response = self.get('accounts', params=params) + return response['data'] + + def get_account(self, account_id, **params): + response = self.get('accounts/%s' % account_id, params=params) + return response['data'] + + def get_funding_instruments(self, account_id, **params): + response = self.get('accounts/%s/funding_instruments' % account_id, params=params) + 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) + return response['data'] + + def get_campaigns(self, account_id, **params): + response = self.get('accounts/%s/campaigns' % account_id, params) + return response['data'] + + def create_campaign(self, account_id, **params): + response = self.post('accounts/%s/campaigns' % account_id, params) + return response['data']