Spaces:
Running
Running
| from dataclasses import dataclass | |
| import requests | |
| class User: | |
| name: str | |
| organizations: list[str] | |
| 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 | |