Merged Twitter Ads functionality with the Twitter functionality.

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)

tests/test_endpoints_ads.py

@@ -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)

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 import EndpointsMixin, EndpointsAdsMixin
-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

twython/endpoints.py

@@ -1,19 +1,5 @@
 # -*- coding: utf-8 -*-
 
-"""
-twython.endpoints
-~~~~~~~~~~~~~~~~~
-
-This module provides a mixin for a :class:`Twython <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 <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 <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

twython/endpoints_ads.py

@@ -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 <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