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 | |
| }} | |
| }} | |
| }} | |
| }} | |
| }} | |
| }} | |
| """ | |
| 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 | |
| deletions | |
| additions | |
| author {{ | |
| user {{ | |
| login | |
| organizations(first: 100) {{ | |
| nodes {{ | |
| name | |
| }} | |
| }} | |
| }} | |
| }} | |
| }} | |
| 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=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 | |