Spaces:
Running
Running
File size: 3,581 Bytes
d80253f b9b07ad d80253f b9b07ad d80253f b9b07ad d80253f b9b07ad d80253f b9b07ad d80253f b9b07ad d80253f |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 |
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
}}
}}
}}
}}
}}
}}
"""
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
|