graphql_calls.py

from dataclasses import dataclass

import requests


@dataclass
class User:
    name: str
    organizations: list[str]


@dataclass
class Commit:
    message: str
    user: User
    additions: int
    deletions: int


def call_with_query(query, token):
    url = "https://api.github.com/graphql"
    r = requests.post(url, json={"query": query}, headers={"Authorization": f"Bearer {token}"})
    return r.json()


def get_tag_commit_date(token, repository, tag):
    owner, name = repository.split("/")
    query = f"""
    query GetTagCommit {{
        repository(owner: "{owner}", name: "{name}"){{
            object(expression: "{tag}") {{
                ... on Commit {{
                    oid
                    message
                    committedDate
                    author {{
                        user {{
                            login
                        }}
                    }}
                }}
            }}
        }}
    }}
    """

    response = call_with_query(query, token)

    try:
        repository = response["data"]["repository"]["object"]

        if repository is None:
            if "errors" in response:
                raise ValueError(response["errors"][0]["message"])
            raise ValueError("Invalid tag. Does this tag exist?")

        committed_date = repository["committedDate"]
    except (KeyError, TypeError):
        raise ValueError("Invalid token. Does your token have the valid permissions?")

    return committed_date


def get_commits(token, repository, branch, since):
    owner, name = repository.split("/")

    def get_page_result(next_page=""):
        query = f"""
        query GetCommits {{
            repository(owner: "{owner}", name: "{name}"){{
                nameWithOwner
                object(expression: "{branch}") {{
                    ... on Commit {{
                        oid
                        history(first: 100, since: "{since}"{next_page}) {{
                            nodes {{
                                message
                                deletions
                                additions
                                author {{
                                    user {{
                                        login
                                        organizations(first: 100) {{
                                            nodes {{
                                                name
                                            }}
                                        }}
                                    }}
                                }}
                            }}
                            pageInfo {{
                                endCursor
                                hasNextPage
                            }}
                        }}
                    }}
                }}
            }}
        }}
        """
        result = call_with_query(query, token)

        if "data" not in result:
            raise ValueError(result["errors"][0]["message"])

        if result["data"]["repository"]["object"] is None:
            raise ValueError("Either the tag or the branch were incorrect.")

        nodes = result["data"]["repository"]["object"]["history"]["nodes"]
        cursor = result["data"]["repository"]["object"]["history"]["pageInfo"]["endCursor"]
        return nodes, cursor

    nodes, cursor = get_page_result()

    while cursor is not None:
        _nodes, cursor = get_page_result(f' after:"{cursor}"')
        nodes.extend(_nodes)

    commits = []
    for node in nodes:
        if node["author"]["user"] is None:
            commits.append(
                Commit(
                    message=node["message"].split("\n")[0],
                    user=User(name="<NOT FOUND>", organizations=[]),
                    additions=node.get("additions"),
                    deletions=node.get("deletions"),
                )
            )
        else:
            commits.append(
                Commit(
                    message=node["message"].split("\n")[0],
                    user=User(
                        name=node["author"]["user"]["login"],
                        organizations=[n["name"] for n in node["author"]["user"]["organizations"]["nodes"]],
                    ),
                    additions=node.get("additions"),
                    deletions=node.get("deletions"),
                )
            )

    return commits