For full archive search, sleep in between pages to avoid the 1 request/second limit.
@@ -10,6 +10,7 @@ import json
import logging
import requests
import datetime
+import time
from oauthlib.oauth2 import BackendApplicationClient
from requests.exceptions import ConnectionError
@@ -106,7 +107,8 @@ class Twarc2:
self.connect()
def _search(
- self, url, query, since_id, until_id, start_time, end_time, max_results
+ self, url, query, since_id, until_id, start_time, end_time, max_results,
+ sleep_between=0
):
params = expansions.EVERYTHING.copy()
@@ -123,11 +125,20 @@ class Twarc2:
params["end_time"] = _ts(end_time)
count = 0
+ made_call = time.monotonic()
+
for response in self.get_paginated(url, params=params):
# can return without 'data' if there are no results
if 'data' in response:
count += len(response['data'])
yield response
+
+ # Calculate the amount of time to sleep, accounting for any
+ # processing time used by the rest of the application.
+ time.sleep(
+ max(0, sleep_between - (time.monotonic() - made_call))
+ )
+ made_call = time.monotonic()
else:
log.info(f'no more results for search')
@@ -193,7 +204,8 @@ class Twarc2:
"""
url = "https://api.twitter.com/2/tweets/search/all"
return self._search(
- url, query, since_id, until_id, start_time, end_time, max_results
+ url, query, since_id, until_id, start_time, end_time, max_results,
+ sleep_between=1.05
)
def tweet_lookup(self, tweet_ids):
