feat: Avoid API Rate-Limiting

issue_metrics.py

@@ -22,9 +22,11 @@
 
 import shutil
 import sys
+from time import sleep
 from typing import List, Union
 
 import github3
+import github3.structs
 from auth import auth_to_github, get_github_app_installation_token
 from classes import IssueWithMetrics
 from config import EnvVars, get_env_vars
@@ -62,19 +64,49 @@ def search_issues(
     Returns:
         List[github3.search.IssueSearchResult]: A list of issues that match the search query.
     """
+
+    # Rate Limit Handling: API only allows 30 requests per minute
+    def wait_for_api_refresh(iterator: github3.structs.SearchIterator):
+        max_retries = 5
+        retry_count = 0
+        sleep_time = 70
+
+        while iterator.ratelimit_remaining < 5:
+            if retry_count >= max_retries:
+                raise RuntimeError("Exceeded maximum retries for API rate limit")
+
+            print(
+                f"GitHub API Rate Limit Low, waiting {sleep_time} seconds to refresh."
+            )
+            sleep(sleep_time)
+
+            # Exponentially increase the sleep time for the next retry
+            sleep_time *= 2
+            retry_count += 1
+
+    issues_per_page = 100
+
     print("Searching for issues...")
-    issues_iterator = github_connection.search_issues(search_query, per_page=100)
+    issues_iterator = github_connection.search_issues(
+        search_query, per_page=issues_per_page
+    )
+    wait_for_api_refresh(issues_iterator)
 
-    # Print the issue titles
     issues = []
     repos_and_owners_string = ""
     for item in owners_and_repositories:
         repos_and_owners_string += f"{item['owner']}/{item['repository']} "
 
+    # Print the issue titles
     try:
-        for issue in issues_iterator:
+        for idx, issue in enumerate(issues_iterator, 1):
             print(issue.title)  # type: ignore
             issues.append(issue)
+
+            # requests are sent once per page of issues
+            if idx % issues_per_page == 0:
+                wait_for_api_refresh(issues_iterator)
+
     except github3.exceptions.ForbiddenError:
         print(
             f"You do not have permission to view a repository from: '{repos_and_owners_string}'; Check your API Token."

test_issue_metrics.py

@@ -42,13 +42,20 @@ class TestSearchIssues(unittest.TestCase):
 
     def test_search_issues(self):
         """Test that search_issues returns the correct issues."""
+
         # Set up the mock GitHub connection object
-        mock_connection = MagicMock()
         mock_issues = [
             MagicMock(title="Issue 1"),
             MagicMock(title="Issue 2"),
         ]
-        mock_connection.search_issues.return_value = mock_issues
+
+        # simulating github3.structs.SearchIterator return value
+        mock_search_result = MagicMock()
+        mock_search_result.__iter__.return_value = iter(mock_issues)
+        mock_search_result.ratelimit_remaining = 30
+
+        mock_connection = MagicMock()
+        mock_connection.search_issues.return_value = mock_search_result
 
         # Call search_issues and check that it returns the correct issues
         repo_with_owner = {"owner": "owner1", "repository": "repo1"}