mirror of
https://github.com/carbon-language/carbon-lang.git
synced 2026-09-24 11:10:11 +01:00
Dana Jansens
233a58fcb1
Mostly these errors were around `dict` missing arguments, and they are almost always `[str, Any]`. But a real thorn here was `xml.etree.ElementTree.Element`. `ty` insists that this is a generic type, and indeed it appears to be one, or becoming one, in some python version. But it is not generic in python 3.12. So we are stuck in an unsolvable land where: - `ty` gives an error unless you write `[str]` on the type, because it thinks it is generic. - python3.12 gives an error if you do write `[str]` on the type, because it thinks it is not generic. Forcing `ty` to target exactly python 3.12 does not help. So I have just used a linter-ignore comment on that line.
118 lines
3.9 KiB
Python
118 lines
3.9 KiB
Python
"""GitHub GraphQL helpers.
|
|
|
|
https://developer.github.com/v4/explorer/ is very useful for building queries.
|
|
"""
|
|
|
|
__copyright__ = """
|
|
Part of the Carbon Language project, under the Apache License v2.0 with LLVM
|
|
Exceptions. See /LICENSE for license information.
|
|
SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
|
|
"""
|
|
|
|
import argparse
|
|
import os
|
|
from collections.abc import Generator
|
|
from typing import Any, Optional, cast
|
|
|
|
# https://pypi.org/project/gql/
|
|
import gql
|
|
import gql.transport.requests
|
|
|
|
_ENV_TOKEN = "GITHUB_ACCESS_TOKEN"
|
|
|
|
# Query elements for pagination.
|
|
PAGINATION = """pageInfo {
|
|
hasNextPage
|
|
endCursor
|
|
}
|
|
totalCount"""
|
|
|
|
|
|
def add_access_token_arg(
|
|
parser: argparse.ArgumentParser, permissions: str
|
|
) -> None:
|
|
"""Adds a flag to set the access token."""
|
|
access_token = os.environ.get(_ENV_TOKEN, default=None)
|
|
parser.add_argument(
|
|
"--access-token",
|
|
metavar="ACCESS_TOKEN",
|
|
default=access_token,
|
|
required=not access_token,
|
|
help="The access token for use with GitHub. May also be specified in "
|
|
"the environment as %s. The access token should have permissions: %s"
|
|
% (_ENV_TOKEN, permissions),
|
|
)
|
|
|
|
|
|
class Client:
|
|
"""A GitHub GraphQL client."""
|
|
|
|
def __init__(self, parsed_args: argparse.Namespace):
|
|
"""Connects to GitHub."""
|
|
transport = gql.transport.requests.RequestsHTTPTransport(
|
|
url="https://api.github.com/graphql",
|
|
headers={"Authorization": "bearer %s" % parsed_args.access_token},
|
|
)
|
|
self._client = gql.Client(transport=transport)
|
|
|
|
def execute(
|
|
self, query: str, variable_values: Optional[dict[str, Any]] = None
|
|
) -> dict[str, Any]:
|
|
"""Runs a query."""
|
|
return cast(
|
|
dict[str, Any],
|
|
self._client.execute(
|
|
gql.gql(query), variable_values=variable_values
|
|
),
|
|
)
|
|
|
|
def execute_and_paginate(
|
|
self,
|
|
query: str,
|
|
path: tuple[str, ...],
|
|
first_page: Optional[dict[str, Any]] = None,
|
|
) -> Generator[dict[str, Any], None, None]:
|
|
"""Runs a query with pagination.
|
|
|
|
Arguments:
|
|
query: The GraphQL query template, which must have both 'cursor' and
|
|
'pagination' fields to fill in. The cursor should be part of the
|
|
location query (with 'first'), and the pagination should be at the
|
|
same level as nodes.
|
|
path: A list of strings indicating the path to the nodes in the
|
|
result.
|
|
first_page: An optional object for the first page of results, which
|
|
will otherwise automatically be collected. This exists for callers
|
|
to optimize by collecting other data with the first page.
|
|
"""
|
|
format = {"cursor": "", "pagination": PAGINATION}
|
|
count = 0
|
|
exp_count = None
|
|
while True:
|
|
if first_page:
|
|
result = first_page
|
|
first_page = None
|
|
else:
|
|
result = self.execute(query % format)
|
|
# Follow the path to the nodes being paginated.
|
|
node_parent = result
|
|
for entry in path:
|
|
node_parent = node_parent[entry]
|
|
# Store the total count of responses.
|
|
if not exp_count:
|
|
exp_count = node_parent["totalCount"]
|
|
# Yield each node individually.
|
|
for node in node_parent["nodes"]:
|
|
yield node
|
|
count += 1
|
|
# Check for pagination, verifying the total count on exit.
|
|
page_info = node_parent["pageInfo"]
|
|
if not page_info["hasNextPage"]:
|
|
assert exp_count == count, "exp %d != actual %d at path %s" % (
|
|
exp_count,
|
|
count,
|
|
path,
|
|
)
|
|
return
|
|
format["cursor"] = ' after: "%s"' % page_info["endCursor"]
|