Make a GitHub API request with retry on transient failures (5xx, 429).
(
method: str,
url: str,
headers: Dict,
timeout: int = None,
max_retries: int = 3,
**kwargs,
)
| 91 | |
| 92 | |
| 93 | def github_api_request( |
| 94 | method: str, |
| 95 | url: str, |
| 96 | headers: Dict, |
| 97 | timeout: int = None, |
| 98 | max_retries: int = 3, |
| 99 | **kwargs, |
| 100 | ) -> requests.Response: |
| 101 | """Make a GitHub API request with retry on transient failures (5xx, 429).""" |
| 102 | _timeout = timeout or DEFAULT_TIMEOUT |
| 103 | resp = None |
| 104 | last_exc = None |
| 105 | for attempt in range(max_retries): |
| 106 | try: |
| 107 | resp = requests.request(method, url, headers=headers, timeout=_timeout, **kwargs) |
| 108 | if resp.status_code < 500 and resp.status_code != 429: |
| 109 | return resp |
| 110 | print(f" GitHub API {resp.status_code} on attempt {attempt + 1}/{max_retries}") |
| 111 | except requests.exceptions.RequestException as e: |
| 112 | last_exc = e |
| 113 | print(f" GitHub API request error on attempt {attempt + 1}/{max_retries}: {e}") |
| 114 | if attempt < max_retries - 1: |
| 115 | delay = INITIAL_RETRY_DELAY * (2 ** attempt) |
| 116 | time.sleep(delay) |
| 117 | # Exhausted retries — raise on network error, return 5xx response for callers to handle |
| 118 | if last_exc: |
| 119 | raise last_exc |
| 120 | print(f" WARNING: GitHub API returned {resp.status_code} after {max_retries} retries: {url}") |
| 121 | return resp |
| 122 | |
| 123 | |
| 124 | def parse_json_response(response: str) -> Optional[Dict]: |
no test coverage detected