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