twitter: add block list ids support

for #40, snarfed/bridgy#473

granary/test/test_twitter.py

@@ -19,6 +19,7 @@
 from granary import source
 from granary import twitter
 from granary.twitter import (
+  API_BLOCK_IDS,
   API_BLOCKS,
   API_FAVORITES,
   API_LIST_TIMELINE,
@@ -1122,6 +1123,18 @@ def test_get_blocklist(self):
     self.assert_equals([ACTOR, ACTOR_2, ACTOR_3],
                        self.twitter.get_blocklist())
 
+  def test_get_blocklist_ids(self):
+    self.expect_urlopen(API_BLOCK_IDS % '-1', {
+      'ids': ['1', '2'],
+      'next_cursor_str': '9',
+    })
+    self.expect_urlopen(API_BLOCK_IDS % '9', {
+      'ids': ['4', '5'],
+      'next_cursor_str': '0',
+    })
+    self.mox.ReplayAll()
+    self.assert_equals(['1', '2', '4', '5'], self.twitter.get_blocklist_ids())
+
   def test_tweet_to_activity_full(self):
     self.assert_equals(ACTIVITY, self.twitter.tweet_to_activity(TWEET))
 

granary/twitter.py

@@ -36,6 +36,7 @@
 from oauth_dropins.webutil import util
 
 API_BASE = 'https://api.twitter.com/1.1/'
+API_BLOCK_IDS = 'blocks/ids.json?count=5000&stringify_ids=true&cursor=%s'
 API_BLOCKS = 'blocks/list.json?skip_status=true&count=5000&cursor=%s'
 API_CURRENT_USER = 'account/verify_credentials.json'
 API_FAVORITES = 'favorites/list.json?screen_name=%s&include_entities=true&tweet_mode=extended'
@@ -142,6 +143,29 @@ class Twitter(source.Source):
   </blockquote>
   """
 
+  def get_blocklist_ids(self):
+    """Returns the current user's block list as a list of Twitter user ids.
+
+    May make multiple API calls, using cursors, to fully fetch large blocklists.
+    https://dev.twitter.com/overview/api/cursoring
+
+    Subject to the same rate limiting as get_blocklist(), but each API call
+    returns ~4k ids, so realistically this can actually fetch entire large
+    blocklists at once.
+
+    Returns:
+      sequence of string Twitter user ids
+    """
+    ids = []
+    cursor = '-1'
+    while cursor and cursor != '0':
+      resp = self.urlopen(API_BLOCK_IDS % cursor)
+      logging.info(json.dumps(resp, indent=2))
+      ids.extend(resp.get('ids', []))
+      cursor = resp.get('next_cursor_str')
+
+    return ids
+
   URL_CANONICALIZER = util.UrlCanonicalizer(
     domain=DOMAIN,
     approve=TWEET_URL_RE,