mirror of
https://github.com/carbon-language/carbon-lang.git
synced 2026-09-24 22:02:23 +01:00
Chandler Carruth
824d7e3c83
This should detect when a PR has a dependency of another open PR and add a comment and label describing it. The comment will even do a best-effort to compute the best starting commit for review. Whenever PRs are closed, it will also scan the open depnedent PRs and try to either remove the PRs in the comment or if it reaches zero the label. It works to update a single comment on a PR rather than adding more comments. Assisted-by: Antigravity with Gemini
118 lines
3.8 KiB
Python
118 lines
3.8 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
|
|
from collections.abc import Generator
|
|
import os
|
|
from typing import Optional, cast
|
|
|
|
# https://pypi.org/project/gql/
|
|
import gql # type: ignore
|
|
import gql.transport.requests # type: ignore
|
|
|
|
_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] = None
|
|
) -> dict:
|
|
"""Runs a query."""
|
|
return cast(
|
|
dict,
|
|
self._client.execute(
|
|
gql.gql(query), variable_values=variable_values
|
|
),
|
|
)
|
|
|
|
def execute_and_paginate(
|
|
self,
|
|
query: str,
|
|
path: tuple[str, ...],
|
|
first_page: Optional[dict] = None,
|
|
) -> Generator[dict, 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"]
|