File size: 4,430 Bytes
3bbc731 b8f978e 3bbc731 b8f978e 3bbc731 f30c6d8 3bbc731 f30c6d8 3bbc731 e326eb8 f30c6d8 2455fd8 f30c6d8 2455fd8 f30c6d8 2455fd8 f30c6d8 e326eb8 3bbc731 a9f33a2 f30c6d8 3bbc731 f30c6d8 3bbc731 a9f33a2 3bbc731 b8f978e 3bbc731 b8f978e 3bbc731 a9f33a2 f30c6d8 2455fd8 f30c6d8 a9f33a2 f30c6d8 3bbc731 f30c6d8 3bbc731 f30c6d8 3bbc731 |
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 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 |
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
|