Rate Limit Countdown for the API Module

pytwitter/api.py

@@ -23,6 +23,9 @@
 from pytwitter.utils.validators import enf_comma_separated
 
 logger = logging.getLogger(__name__)
+logging.basicConfig(level=logging.INFO,
+                    format='%(asctime)s:%(name)s:%(levelname)s: %(message)s',
+                    datefmt='%Y-%m-%d %H:%M:%S')
 
 
 class Api:
@@ -47,6 +50,7 @@ def __init__(
         application_only_auth: bool = False,
         oauth_flow: bool = False,  # provide access with authorize
         sleep_on_rate_limit: bool = False,
+        notify_on_rate_limit: bool = False,
         timeout: Optional[int] = None,
         proxies: Optional[dict] = None,
         callback_uri: Optional[str] = None,
@@ -66,6 +70,7 @@ def __init__(
         :param application_only_auth: If set this, with auto exchange app bearer token with consumer credentials.
         :param oauth_flow: If set this, You need generate access token with user by OAuth1.1 or OAuth2.0
         :param sleep_on_rate_limit: If token reach the limit, will sleep.
+        :param notify_on_rate_limit: Provides visual feedback on remaining rate limit sleep period.
         :param timeout: Timeout for requests.
         :param proxies: Proxies for requests.
         :param callback_uri: Your callback URL. This value must correspond to one of the Callback URLs defined in your App settings.
@@ -82,6 +87,7 @@ def __init__(
         self.proxies = proxies
         self.rate_limit = RateLimit()
         self.sleep_on_rate_limit = sleep_on_rate_limit
+        self.notify_on_rate_limit = notify_on_rate_limit
         self.auth_user_id = None  # Note: use this keep uid for auth user
         self.callback_uri = (
             callback_uri if callback_uri is not None else self.DEFAULT_CALLBACK_URI
@@ -130,6 +136,33 @@ def get_uid_from_access_token_key(access_token: str) -> str:
         """
         uid, _ = access_token.split("-")
         return uid
+
+    @staticmethod
+    def rate_limit_countdown(seconds):
+        """
+        Provides a visual countdown in seconds when a
+        rate limit threshold has been encountered.
+
+        :param seconds: number of seconds to sleep
+        :return: None
+        """
+        print('Twitter rate limit threshold encountered.')
+        sleeping_period = int("%02d" % (seconds)) / 60
+        print(f"Connection sleeping for {'%g' % sleeping_period} minutes")
+        start = time.time()
+        time.process_time()
+        elapsed = 0
+        while elapsed < seconds:
+            elapsed = time.time() - start
+            current_time = seconds - int("%02d" % (elapsed))
+            if int('%02d' % (current_time)) == 0:
+                return None
+            elif (int("%02d" % (current_time)) % 60 == 0) and int("%02d" % (current_time)) < int("%02d" % (seconds)):
+                minutes_remaining = int("%02d" % (current_time)) / 60
+                print(f"Remaining time until rate limit threshold resets is: {'%g' % minutes_remaining} minutes")
+                time.sleep(1)
+            else:
+                time.sleep(1)
 
     def _request(
         self, url, verb="GET", params=None, data=None, json=None, enforce_auth=True
@@ -154,11 +187,15 @@ def _request(
             if url and self.sleep_on_rate_limit:
                 limit = self.rate_limit.get_limit(url=url, method=verb)
                 if limit.remaining == 0:
-                    s_time = max((limit.reset - time.time()), 0) + 10.0
-                    logger.debug(
-                        f"Rate limited requesting [{url}], sleeping for [{s_time}]"
-                    )
-                    time.sleep(s_time)
+                    if self.notify_on_rate_limit:
+                        s_time = max((limit.reset - time.time()), 0) + 10.0
+                        logger.debug(f"Rate limited requesting [{url}], sleeping for [{s_time}]")
+                        self.rate_limit_countdown(s_time)
+                        time.sleep(s_time)
+                    else:
+                        s_time = max((limit.reset - time.time()), 0) + 10.0
+                        logger.debug(f"Rate limited requesting [{url}], sleeping for [{s_time}]")
+                        time.sleep(s_time)
 
         resp = self.session.request(
             url=url,