graphql_calls.py

from dataclasses import dataclass

import requests


@dataclass
class Commit:
    message: str
    user: str


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
                        }}
                    }}
                }}
            }}
        }}
    }}
    """
    return call_with_query(query, token)['data']['repository']['object']['committedDate']


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

    def get_page_result(next_page=''):
        query = f"""
        query GetCommits {{
            repository(owner: "{owner}", name: "{name}"){{
                nameWithOwner
                object(expression: "main") {{
                    ... on Commit {{
                        oid
                        history(first: 100, since: "{since}"{next_page}) {{
                            nodes {{
                                message
                                author {{
                                    user {{
                                        login
                                    }}
                                }}
                            }}
                            pageInfo {{
                                endCursor
                                hasNextPage
                            }}
                        }}
                    }}
                }}
            }}
        }}
        """
        result = call_with_query(query, token)
        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='<NOT FOUND>'))
        else:
            commits.append(Commit(message=node['message'].split('\n')[0], user=node['author']['user']['login']))

    return commits