Download JSON files from a GitHub repository folder. Args: repo_owner: GitHub repository owner username repo_name: GitHub repository name folder_path: Local folder path to save downloaded files branch: Git branch name (default: 'main') github_t
(
repo_owner: str,
repo_name: str,
folder_path: str,
branch: str = 'main',
github_token: Optional[str] = None,
delay_between_requests: float = 0.5,
max_retries: int = 5,
retry_delay: float = 60
)
| 5 | from typing import Optional |
| 6 | |
| 7 | def download_github_folder( |
| 8 | repo_owner: str, |
| 9 | repo_name: str, |
| 10 | folder_path: str, |
| 11 | branch: str = 'main', |
| 12 | github_token: Optional[str] = None, |
| 13 | delay_between_requests: float = 0.5, |
| 14 | max_retries: int = 5, |
| 15 | retry_delay: float = 60 |
| 16 | ) -> None: |
| 17 | """Download JSON files from a GitHub repository folder. |
| 18 | |
| 19 | Args: |
| 20 | repo_owner: GitHub repository owner username |
| 21 | repo_name: GitHub repository name |
| 22 | folder_path: Local folder path to save downloaded files |
| 23 | branch: Git branch name (default: 'main') |
| 24 | github_token: GitHub personal access token (optional, from GITHUB_TOKEN env var if not provided) |
| 25 | delay_between_requests: Delay in seconds between API requests to avoid rate limiting (default: 0.5) |
| 26 | max_retries: Maximum number of retries for rate limit errors (default: 5) |
| 27 | retry_delay: Delay in seconds before retrying after rate limit error (default: 60) |
| 28 | """ |
| 29 | # Get token from parameter or environment variable |
| 30 | if github_token is None: |
| 31 | github_token = os.getenv('GITHUB_TOKEN') |
| 32 | |
| 33 | # Prepare headers with token if available |
| 34 | headers = {} |
| 35 | if github_token: |
| 36 | headers['Authorization'] = f'token {github_token}' |
| 37 | |
| 38 | api_url = f"https://api.github.com/repos/{repo_owner}/{repo_name}/contents/key_point?ref={branch}" |
| 39 | |
| 40 | # Helper function to make request with retry logic |
| 41 | def make_request_with_retry(url: str, retry_count: int = 0): |
| 42 | """Make a request with retry logic for rate limiting.""" |
| 43 | response = requests.get(url, headers=headers) |
| 44 | |
| 45 | # Check for rate limit error |
| 46 | if response.status_code == 403: |
| 47 | # Check if it's a rate limit error |
| 48 | rate_limit_remaining = response.headers.get('X-RateLimit-Remaining', '0') |
| 49 | rate_limit_reset = response.headers.get('X-RateLimit-Reset', '0') |
| 50 | |
| 51 | if rate_limit_remaining == '0' and retry_count < max_retries: |
| 52 | reset_time = int(rate_limit_reset) |
| 53 | wait_time = max(retry_delay, reset_time - int(time.time()) + 5) |
| 54 | print(f"Rate limit exceeded. Waiting {wait_time:.0f} seconds before retry {retry_count + 1}/{max_retries}...") |
| 55 | time.sleep(wait_time) |
| 56 | return make_request_with_retry(url, retry_count + 1) |
| 57 | else: |
| 58 | response.raise_for_status() |
| 59 | |
| 60 | response.raise_for_status() |
| 61 | return response |
| 62 | |
| 63 | # Get folder contents |
| 64 | response = make_request_with_retry(api_url) |
no test coverage detected