Fetches and returns all open issues and pull requests from the `internetarchive/openlibrary` repository. GitHub API results are paginated. This functions appends each result to a list, and does so for all pages. To keep API calls to a minimum, we request the maximum number of results
()
| 38 | |
| 39 | |
| 40 | def fetch_issues(): |
| 41 | """ |
| 42 | Fetches and returns all open issues and pull requests from the `internetarchive/openlibrary` repository. |
| 43 | |
| 44 | GitHub API results are paginated. This functions appends each result to a list, and does so for all pages. |
| 45 | To keep API calls to a minimum, we request the maximum number of results per request (100 per page, as of writing). |
| 46 | |
| 47 | Calls to fetch issues from Github are considered critical, and a failure of any such call will cause the script to |
| 48 | fail fast. |
| 49 | """ |
| 50 | # Make initial query for open issues: |
| 51 | p = {"state": "open", "per_page": 100} |
| 52 | response = requests.get( |
| 53 | "https://api.github.com/repos/internetarchive/openlibrary/issues", |
| 54 | params=p, |
| 55 | headers=github_headers, |
| 56 | ) |
| 57 | d = response.json() |
| 58 | if response.status_code != 200: |
| 59 | print("Initial request for issues has failed.") |
| 60 | print(f"Message: {d.get('message', '')}") |
| 61 | print(f"Documentation URL: {d.get('documentation_url', '')}") |
| 62 | response.raise_for_status() |
| 63 | |
| 64 | results = d |
| 65 | |
| 66 | # Fetch additional updated issues, if any exist |
| 67 | def get_next_page(url: str): |
| 68 | """Returns list of issues and optional url for next page""" |
| 69 | # Get issues |
| 70 | resp = requests.get(url, headers=github_headers) |
| 71 | d = resp.json() |
| 72 | |
| 73 | if resp.status_code != 200: |
| 74 | print("Request for next page of issues has failed.") |
| 75 | print(f"Message: {d.get('message', '')}") |
| 76 | print(f"Documentation URL: {d.get('documentation_url', '')}") |
| 77 | response.raise_for_status() |
| 78 | |
| 79 | issues = d |
| 80 | |
| 81 | # Prepare url for next page |
| 82 | next = resp.links.get("next", {}) |
| 83 | next_url = next.get("url", "") |
| 84 | |
| 85 | return issues, next_url |
| 86 | |
| 87 | links = response.links |
| 88 | next = links.get("next", {}) |
| 89 | next_url = next.get("url", "") |
| 90 | while next_url: |
| 91 | # Wait one second... |
| 92 | time.sleep(1) |
| 93 | # ...then, make call for more issues with next link |
| 94 | issues, next_url = get_next_page(next_url) |
| 95 | results = results + issues |
| 96 | |
| 97 | return results |
no test coverage detected