| 40 | self._session.headers['accept'] = 'application/vnd.github.v3+json' |
| 41 | |
| 42 | def api_request(self, method: str, *args, **kwargs) -> requests.Response: |
| 43 | response = self._session.request(method, *args, **kwargs) |
| 44 | debug(f'{response.request.method} {response.request.url} -> {response.status_code}') |
| 45 | |
| 46 | if 400 <= response.status_code < 500: |
| 47 | try: |
| 48 | message = response.json()['message'] |
| 49 | |
| 50 | if response.headers['X-RateLimit-Remaining'] == '0' and response.headers['X-RateLimit-Limit'] != '0': |
| 51 | limit_reset = datetime.datetime.fromtimestamp(int(response.headers['X-RateLimit-Reset'])) |
| 52 | sys.stdout.write(message) |
| 53 | sys.stdout.write(f' Try again when the rate limit resets at {limit_reset} UTC.\n') |
| 54 | sys.exit(1) |
| 55 | |
| 56 | if message not in ['Resource not accessible by integration', 'Personal access tokens with fine grained access do not support the GraphQL API']: |
| 57 | sys.stdout.write(message) |
| 58 | sys.stdout.write('\n') |
| 59 | debug(response.content.decode()) |
| 60 | |
| 61 | except Exception: |
| 62 | sys.stdout.write(response.content.decode()) |
| 63 | sys.stdout.write('\n') |
| 64 | raise |
| 65 | |
| 66 | return response |
| 67 | |
| 68 | def get(self, path: str, **kwargs: Any) -> Response: |
| 69 | return self.api_request('GET', path, **kwargs) |