(token: str, repo: str)
| 50 | |
| 51 | |
| 52 | def _search_issues_graphql(token: str, repo: str) -> list[dict[str, Any]]: |
| 53 | query = """ |
| 54 | query($searchQuery: String!, $cursor: String) { |
| 55 | search(query: $searchQuery, type: ISSUE, first: 100, after: $cursor) { |
| 56 | issueCount |
| 57 | pageInfo { |
| 58 | hasNextPage |
| 59 | endCursor |
| 60 | } |
| 61 | nodes { |
| 62 | ... on Issue { |
| 63 | number |
| 64 | title |
| 65 | body |
| 66 | url |
| 67 | state |
| 68 | repository { |
| 69 | nameWithOwner |
| 70 | } |
| 71 | } |
| 72 | } |
| 73 | } |
| 74 | } |
| 75 | """ |
| 76 | |
| 77 | search_query = f'repo:{repo} type:issue in:body "ci-regexp:"' |
| 78 | all_issues: list[dict[str, Any]] = [] |
| 79 | cursor = None |
| 80 | |
| 81 | while True: |
| 82 | variables: dict[str, Any] = {"searchQuery": search_query} |
| 83 | if cursor: |
| 84 | variables["cursor"] = cursor |
| 85 | |
| 86 | response = requests.post( |
| 87 | "https://api.github.com/graphql", |
| 88 | headers={ |
| 89 | "Authorization": f"Bearer {token}", |
| 90 | "Content-Type": "application/json", |
| 91 | }, |
| 92 | json={"query": query, "variables": variables}, |
| 93 | ) |
| 94 | |
| 95 | if response.status_code != 200: |
| 96 | rate_limit = response.headers.get("X-RateLimit-Remaining", "unknown") |
| 97 | rate_limit_reset = response.headers.get("X-RateLimit-Reset", "unknown") |
| 98 | raise ValueError( |
| 99 | f"Bad return code from GitHub GraphQL: {response.status_code}, " |
| 100 | f"rate_limit_remaining={rate_limit}, " |
| 101 | f"rate_limit_reset={rate_limit_reset}, " |
| 102 | f"response={response.text[:500]}, " |
| 103 | f"has_token=True" |
| 104 | ) |
| 105 | |
| 106 | result = response.json() |
| 107 | if "errors" in result: |
| 108 | raise ValueError(f"GitHub GraphQL errors: {result['errors']}") |
| 109 |
no test coverage detected