(self, query: str)
| 311 | self.headers['Authorization'] = f'bearer {token}' |
| 312 | |
| 313 | def _request_graphql(self, query: str) -> dict[str, object]: |
| 314 | last_error: Exception | None = None |
| 315 | for attempt in range(3): |
| 316 | try: |
| 317 | payload = json.dumps({'query': query}).encode() |
| 318 | http_request = request.Request( |
| 319 | 'https://api.github.com/graphql', |
| 320 | data=payload, |
| 321 | headers={ |
| 322 | **self.headers, |
| 323 | 'Content-Type': 'application/json', |
| 324 | }, |
| 325 | method='POST', |
| 326 | ) |
| 327 | with request.urlopen(http_request, timeout=60) as response: |
| 328 | payload = json.loads(response.read().decode()) |
| 329 | if payload.get('errors'): |
| 330 | raise RuntimeError( |
| 331 | f"GitHub GraphQL query failed: {payload['errors']}" |
| 332 | ) |
| 333 | return payload |
| 334 | except error.HTTPError as http_error: |
| 335 | last_error = http_error |
| 336 | if http_error.code in RETRYABLE_STATUS_CODES and attempt < 2: |
| 337 | time.sleep(1 + attempt) |
| 338 | continue |
| 339 | raise |
| 340 | except (error.URLError, ValueError, RuntimeError) as request_error: |
| 341 | last_error = request_error |
| 342 | if attempt == 2: |
| 343 | raise |
| 344 | time.sleep(1 + attempt) |
| 345 | assert last_error is not None |
| 346 | raise last_error |
| 347 | |
| 348 | def _fetch_batch(self, batch: list[int]) -> dict[int, PullRequestRecord]: |
| 349 | query_parts = [] |
no test coverage detected