diff --git a/tests/test_endpoints.py b/tests/test_endpoints.py index aa79998..28cf68b 100644 --- a/tests/test_endpoints.py +++ b/tests/test_endpoints.py @@ -1,15 +1,17 @@ +import base64 +import datetime +import urllib +import time 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, unittest + access_token, test_tweet_object, test_tweet_html, unittest, + test_account_id, test_funding_instrument_id, test_campaign_id ) -import time - - class TwythonEndpointsTestCase(unittest.TestCase): def setUp(self): @@ -531,3 +533,261 @@ class TwythonEndpointsTestCase(unittest.TestCase): def test_get_application_rate_limit_status(self): """Test getting application rate limit status succeeds""" self.oauth2_api.get_application_rate_limit_status() + + +class TwythonEndpointsAdsTestCase(unittest.TestCase): + TEST_CAMPAIGN = { + 'name': 'Test Twitter campaign - Twython', + 'funding_instrument_id': test_funding_instrument_id, + 'start_time': datetime.datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ'), + 'daily_budget_amount_local_micro': 10 * 1000000, + 'paused': True + } + + TEST_WEBSITE_CLICKS_LINE_ITEM = { + 'bid_type': 'MAX', + 'bid_amount_local_micro': 2000000, + 'product_type': 'PROMOTED_TWEETS', + 'placements': 'ALL_ON_TWITTER', + 'objective': 'WEBSITE_CLICKS', + 'paused': True + } + + 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 = 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') + def test_get_accounts(self): + accounts = self.api.get_accounts() + self.assertTrue(len(accounts) >= 0) + + @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') + 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') + 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') + 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') + + @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') + 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') + def test_get_available_locations(self): + params = { + 'location_type': 'CITY', + 'country_code': 'US' + } + available_locations = self.api.get_available_locations(**params) + self.assertTrue(len(available_locations) > 0) + + @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') + def test_create_and_delete_campaign(self): + campaign_id = self._create_test_campaign() + campaign_check = self.api.get_campaign(test_account_id, campaign_id) + self.assertEqual(campaign_check['id'], campaign_id) + self._delete_test_campaign(campaign_id) + + def _create_test_campaign(self): + campaign = self.api.create_campaign(test_account_id, **self.TEST_CAMPAIGN) + campaign_id = campaign['id'] + self.assertEqual(campaign['account_id'], test_account_id) + self.assertIsNotNone(campaign_id) + return campaign_id + + def _delete_test_campaign(self, campaign_id): + is_deleted = self.api.delete_campaign(test_account_id, campaign_id) + self.assertTrue(is_deleted) + + @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) + line_items = self.api.get_line_items(test_account_id, campaign_id) + self.assertTrue(len(line_items) > 0) + self._delete_test_line_item(line_item_id) + self._delete_test_campaign(campaign_id) + + def _create_test_line_item(self, campaign_id): + response = self.api.create_line_item(test_account_id, campaign_id, **self.TEST_WEBSITE_CLICKS_LINE_ITEM) + line_item_id = response['id'] + self.assertEqual(response['account_id'], test_account_id) + self.assertEqual(response['campaign_id'], campaign_id) + self.assertIsNotNone(line_item_id) + return line_item_id + + def _delete_test_line_item(self, line_item_id): + is_deleted = self.api.delete_line_item(test_account_id, line_item_id) + self.assertTrue(is_deleted) + + @unittest.skip('skipping non-updated test') + def test_upload_image(self): + response = self._upload_test_image() + self.assertIsNotNone(response['media_id']) + + def _upload_test_image(self): + image_file = urllib.urlopen('https://openclipart.org/image/800px/svg_to_png/190042/1389527622.png').read() + image_file_encoded = base64.b64encode(image_file) + upload_data = { + 'media_data': image_file_encoded + # the line below will have to be provided once we start uploading photos on behalf of advertisers + # 'additional_owners': '' + } + response = self.api.upload_image(**upload_data) + return response + + @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') + 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) + self.assertEqual(card['id'], card_id) + self._delete_test_website_card(card_id) + + def _create_test_website_card(self): + uploaded_image = self._upload_test_image() + test_website_card = { + 'name': 'Zemanta Partnered with AdsNative for Programmatic Native Supply', + 'website_title': 'Zemanta Partnered with AdsNative for Programmatic Native Supply', + 'website_url': 'http://r1.zemanta.com/r/u1tllsoizjls/facebook/1009/92325/', + 'website_cta': 'READ_MORE', + 'image_media_id': uploaded_image['media_id_string'] + } + response_create = self.api.create_website_card(test_account_id, **test_website_card) + card_id = response_create['id'] + self.assertEqual(response_create['account_id'], test_account_id) + self.assertIsNotNone(card_id) + return card_id + + def _delete_test_website_card(self, card_id): + 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') + def test_create_promoted_only_tweet(self): + card_id, tweet_id = self._create_test_promoted_only_tweet() + self._delete_test_website_card(card_id) + + def _create_test_promoted_only_tweet(self): + card_id = self._create_test_website_card() + card = self.api.get_website_card(test_account_id, card_id) + test_promoted_only_tweet = { + 'status': 'This is test tweet for website card: %s' % card['preview_url'], + # 'as_user_id': '', + } + response = self.api.create_promoted_only_tweet(test_account_id, **test_promoted_only_tweet) + tweet_id = response['id'] + self.assertIsNotNone(tweet_id) + return card_id, tweet_id + + @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) + card_id, tweet_id = self._create_test_promoted_only_tweet() + test_tweet_promotion = { + 'line_item_id': line_item_id, + 'tweet_ids': [tweet_id] + } + result_promote = self.api.promote_tweet(test_account_id, **test_tweet_promotion) + self.assertTrue(len(result_promote) > 0) + self.assertEqual(int(result_promote[0]['tweet_id']), tweet_id) + promotion_id = result_promote[0]['id'] + self.assertIsNotNone(promotion_id) + promoted_tweets = self.api.get_promoted_tweets(test_account_id, line_item_id) + self.assertTrue(len(promoted_tweets) == 1) + result_unpromotion = self.api.unpromote_tweet(test_account_id, promotion_id) + self.assertTrue(result_unpromotion['deleted']) + self.assertEqual(result_unpromotion['id'], promotion_id) + self._delete_test_campaign(campaign_id) + self._delete_test_website_card(card_id) + + @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) + criteria_ios_id = self._create_test_targeting_criteria(line_item_id, 'PLATFORM', '0') + criteria_android_id = self._create_test_targeting_criteria(line_item_id, 'PLATFORM', '1') + criteria_desktop_id = self._create_test_targeting_criteria(line_item_id, 'PLATFORM', '4') + criteria_new_york_id = self._create_test_targeting_criteria(line_item_id, 'LOCATION', 'b6c2e04f1673337f') + # since all the targeting criteria share the same id, we only have to do the removal once. + self.api.remove_targeting_criteria(test_account_id, criteria_ios_id) + self.api.remove_targeting_criteria(test_account_id, criteria_android_id) + self.api.remove_targeting_criteria(test_account_id, criteria_desktop_id) + self.api.remove_targeting_criteria(test_account_id, criteria_new_york_id) + self._delete_test_line_item(line_item_id) + self._delete_test_campaign(campaign_id) + + def _create_test_targeting_criteria(self, line_item_id, targeting_type, targeting_value): + test_targeting_criteria_ios = { + 'targeting_type': targeting_type, + 'targeting_value': targeting_value + } + response_add = self.api.add_targeting_criteria(test_account_id, line_item_id, **test_targeting_criteria_ios) + self.assertEqual(response_add['account_id'], test_account_id) + self.assertEquals(response_add['line_item_id'], line_item_id) + return response_add['id'] + + @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']) + promoted_ids = [tweet['id'] for tweet in promoted_tweets] + stats_query = { + 'start_time': '2015-10-29T00:00:00Z', + 'end_time': '2015-10-29T23:59:59Z', + 'granularity': 'TOTAL' + } + stats = self.api.get_stats_promoted_tweets(test_account_id, promoted_ids, **stats_query) + self.assertTrue(len(stats) >= 0) diff --git a/tests/test_endpoints_ads.py b/tests/test_endpoints_ads.py deleted file mode 100644 index 57c0c6c..0000000 --- a/tests/test_endpoints_ads.py +++ /dev/null @@ -1,266 +0,0 @@ -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 -) - - -class TwythonEndpointsTestCase(unittest.TestCase): - TEST_CAMPAIGN = { - 'name': 'Test Twitter campaign - Twython', - 'funding_instrument_id': test_funding_instrument_id, - 'start_time': datetime.datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ'), - 'daily_budget_amount_local_micro': 10 * 1000000, - 'paused': True - } - - TEST_WEBSITE_CLICKS_LINE_ITEM = { - 'bid_type': 'MAX', - 'bid_amount_local_micro': 2000000, - 'product_type': 'PROMOTED_TWEETS', - 'placements': 'ALL_ON_TWITTER', - 'objective': 'WEBSITE_CLICKS', - 'paused': True - } - - 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 = 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') - def test_get_accounts(self): - accounts = self.api.get_accounts() - self.assertTrue(len(accounts) >= 0) - - @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') - 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') - 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') - 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') - - @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') - 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') - def test_get_available_locations(self): - params = { - 'location_type': 'CITY', - 'country_code': 'US' - } - available_locations = self.api.get_available_locations(**params) - self.assertTrue(len(available_locations) > 0) - - @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') - def test_create_and_delete_campaign(self): - campaign_id = self._create_test_campaign() - campaign_check = self.api.get_campaign(test_account_id, campaign_id) - self.assertEqual(campaign_check['id'], campaign_id) - self._delete_test_campaign(campaign_id) - - def _create_test_campaign(self): - campaign = self.api.create_campaign(test_account_id, **self.TEST_CAMPAIGN) - campaign_id = campaign['id'] - self.assertEqual(campaign['account_id'], test_account_id) - self.assertIsNotNone(campaign_id) - return campaign_id - - def _delete_test_campaign(self, campaign_id): - is_deleted = self.api.delete_campaign(test_account_id, campaign_id) - self.assertTrue(is_deleted) - - @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) - line_items = self.api.get_line_items(test_account_id, campaign_id) - self.assertTrue(len(line_items) > 0) - self._delete_test_line_item(line_item_id) - self._delete_test_campaign(campaign_id) - - def _create_test_line_item(self, campaign_id): - response = self.api.create_line_item(test_account_id, campaign_id, **self.TEST_WEBSITE_CLICKS_LINE_ITEM) - line_item_id = response['id'] - self.assertEqual(response['account_id'], test_account_id) - self.assertEqual(response['campaign_id'], campaign_id) - self.assertIsNotNone(line_item_id) - return line_item_id - - def _delete_test_line_item(self, line_item_id): - is_deleted = self.api.delete_line_item(test_account_id, line_item_id) - self.assertTrue(is_deleted) - - @unittest.skip('skipping non-updated test') - def test_upload_image(self): - response = self._upload_test_image() - self.assertIsNotNone(response['media_id']) - - def _upload_test_image(self): - image_file = urllib.urlopen('https://openclipart.org/image/800px/svg_to_png/190042/1389527622.png').read() - image_file_encoded = base64.b64encode(image_file) - upload_data = { - 'media_data': image_file_encoded - # the line below will have to be provided once we start uploading photos on behalf of advertisers - # 'additional_owners': '' - } - response = self.api.upload_image(**upload_data) - return response - - @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') - 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) - self.assertEqual(card['id'], card_id) - self._delete_test_website_card(card_id) - - def _create_test_website_card(self): - uploaded_image = self._upload_test_image() - test_website_card = { - 'name': 'Zemanta Partnered with AdsNative for Programmatic Native Supply', - 'website_title': 'Zemanta Partnered with AdsNative for Programmatic Native Supply', - 'website_url': 'http://r1.zemanta.com/r/u1tllsoizjls/facebook/1009/92325/', - 'website_cta': 'READ_MORE', - 'image_media_id': uploaded_image['media_id_string'] - } - response_create = self.api.create_website_card(test_account_id, **test_website_card) - card_id = response_create['id'] - self.assertEqual(response_create['account_id'], test_account_id) - self.assertIsNotNone(card_id) - return card_id - - def _delete_test_website_card(self, card_id): - 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') - def test_create_promoted_only_tweet(self): - card_id, tweet_id = self._create_test_promoted_only_tweet() - self._delete_test_website_card(card_id) - - def _create_test_promoted_only_tweet(self): - card_id = self._create_test_website_card() - card = self.api.get_website_card(test_account_id, card_id) - test_promoted_only_tweet = { - 'status': 'This is test tweet for website card: %s' % card['preview_url'], - # 'as_user_id': '', - } - response = self.api.create_promoted_only_tweet(test_account_id, **test_promoted_only_tweet) - tweet_id = response['id'] - self.assertIsNotNone(tweet_id) - return card_id, tweet_id - - @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) - card_id, tweet_id = self._create_test_promoted_only_tweet() - test_tweet_promotion = { - 'line_item_id': line_item_id, - 'tweet_ids': [tweet_id] - } - result_promote = self.api.promote_tweet(test_account_id, **test_tweet_promotion) - self.assertTrue(len(result_promote) > 0) - self.assertEqual(int(result_promote[0]['tweet_id']), tweet_id) - promotion_id = result_promote[0]['id'] - self.assertIsNotNone(promotion_id) - promoted_tweets = self.api.get_promoted_tweets(test_account_id, line_item_id) - self.assertTrue(len(promoted_tweets) == 1) - result_unpromotion = self.api.unpromote_tweet(test_account_id, promotion_id) - self.assertTrue(result_unpromotion['deleted']) - self.assertEqual(result_unpromotion['id'], promotion_id) - self._delete_test_campaign(campaign_id) - self._delete_test_website_card(card_id) - - @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) - criteria_ios_id = self._create_test_targeting_criteria(line_item_id, 'PLATFORM', '0') - criteria_android_id = self._create_test_targeting_criteria(line_item_id, 'PLATFORM', '1') - criteria_desktop_id = self._create_test_targeting_criteria(line_item_id, 'PLATFORM', '4') - criteria_new_york_id = self._create_test_targeting_criteria(line_item_id, 'LOCATION', 'b6c2e04f1673337f') - # since all the targeting criteria share the same id, we only have to do the removal once. - self.api.remove_targeting_criteria(test_account_id, criteria_ios_id) - self.api.remove_targeting_criteria(test_account_id, criteria_android_id) - self.api.remove_targeting_criteria(test_account_id, criteria_desktop_id) - self.api.remove_targeting_criteria(test_account_id, criteria_new_york_id) - self._delete_test_line_item(line_item_id) - self._delete_test_campaign(campaign_id) - - def _create_test_targeting_criteria(self, line_item_id, targeting_type, targeting_value): - test_targeting_criteria_ios = { - 'targeting_type': targeting_type, - 'targeting_value': targeting_value - } - response_add = self.api.add_targeting_criteria(test_account_id, line_item_id, **test_targeting_criteria_ios) - self.assertEqual(response_add['account_id'], test_account_id) - self.assertEquals(response_add['line_item_id'], line_item_id) - return response_add['id'] - - @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']) - promoted_ids = [tweet['id'] for tweet in promoted_tweets] - stats_query = { - 'start_time': '2015-10-29T00:00:00Z', - 'end_time': '2015-10-29T23:59:59Z', - 'granularity': 'TOTAL' - } - stats = self.api.get_stats_promoted_tweets(test_account_id, promoted_ids, **stats_query) - self.assertTrue(len(stats) >= 0) diff --git a/twython/api.py b/twython/api.py index add9267..c4bd6f7 100644 --- a/twython/api.py +++ b/twython/api.py @@ -19,8 +19,7 @@ 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 .endpoints import EndpointsMixin -from .endpoints_ads import EndpointsAdsMixin +from .endpoints import EndpointsMixin, EndpointsAdsMixin from .exceptions import TwythonError, TwythonAuthError, TwythonRateLimitError from .helpers import _transparent_params from .api_type import API_TYPE_TWITTER, API_TYPE_TWITTER_ADS diff --git a/twython/endpoints.py b/twython/endpoints.py index 561ec45..cdf1c37 100644 --- a/twython/endpoints.py +++ b/twython/endpoints.py @@ -1,19 +1,5 @@ # -*- coding: utf-8 -*- -""" -twython.endpoints -~~~~~~~~~~~~~~~~~ - -This module provides a mixin for a :class:`Twython ` instance. -Parameters that need to be embedded in the API url just need to be passed -as a keyword argument. - -e.g. Twython.retweet(id=12345) - -This map is organized the order functions are documented at: -https://dev.twitter.com/docs/api/1.1 -""" - import os import warnings try: @@ -22,9 +8,24 @@ except ImportError: from io import StringIO from .advisory import TwythonDeprecationWarning +from .api_type import API_TYPE_TWITTER_ADS class EndpointsMixin(object): + """ + twython.endpoints + ~~~~~~~~~~~~~~~~~ + + This module provides a mixin for a :class:`Twython ` instance. + Parameters that need to be embedded in the API url just need to be passed + as a keyword argument. + + e.g. Twython.retweet(id=12345) + + This map is organized the order functions are documented at: + https://dev.twitter.com/docs/api/1.1 + """ + # Timelines def get_mentions_timeline(self, **params): """Returns the 20 most recent mentions (tweets containing a users's @@ -1058,3 +1059,173 @@ TWITTER_HTTP_STATUS_CODE = { couldn\'t be serviced due to some failure within our stack. Try \ again later.'), } + + +class EndpointsAdsMixin(object): + """ + 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 + """ + def get_accounts(self, **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, 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, 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, + 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, + 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, 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, 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, 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, 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, + 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, 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), 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, + 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), 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, + 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, 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, + 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, 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, + 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, + 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, 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, 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), + 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, + 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, + 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), + 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): + # the promoted_tweet_ids contains a list of up to 20 identifiers: + # https://dev.twitter.com/ads/reference/get/stats/accounts/%3Aaccount_id/promoted_tweets + stats = [] + max_chunk_size = 20 + for i in range(0, len(promoted_tweet_ids), max_chunk_size): + 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, + api_type=API_TYPE_TWITTER_ADS, version=self.api_ads_version) + stats.extend(response['data']) + return stats diff --git a/twython/endpoints_ads.py b/twython/endpoints_ads.py deleted file mode 100644 index 9fe1f3c..0000000 --- a/twython/endpoints_ads.py +++ /dev/null @@ -1,179 +0,0 @@ -# -*- 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 -""" - -from .api_type import API_TYPE_TWITTER_ADS - -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, 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, 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, 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, - 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, - 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, 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, 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, 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, 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, - 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, 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), 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, - 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), 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, - 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, 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, - 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, 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, - 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, - 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, 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, 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), - 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, - 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, - 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), - 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): - # the promoted_tweet_ids contains a list of up to 20 identifiers: - # https://dev.twitter.com/ads/reference/get/stats/accounts/%3Aaccount_id/promoted_tweets - stats = [] - max_chunk_size = 20 - for i in range(0, len(promoted_tweet_ids), max_chunk_size): - 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, - api_type=API_TYPE_TWITTER_ADS, version=self.api_ads_version) - stats.extend(response['data']) - return stats