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
|
import json
import urllib.parse
import urllib.request
from datetime import datetime, timedelta, timezone
from http import HTTPStatus
GITHUB_PER_PAGE = 100
URLOPEN_TIMEOUT_SECONDS = 15
def github_request(method, path, token, repo, data=None, params=None):
"""Make an authenticated GitHub API request."""
url = f"https://api.github.com/repos/{repo}{path}"
if params:
encoded_params = urllib.parse.urlencode(params)
url = f"{url}?{encoded_params}"
headers = {
"Authorization": f"Bearer {token}",
"Accept": "application/vnd.github+json",
"X-GitHub-Api-Version": "2022-11-28",
}
body = None
if data is not None:
body = json.dumps(data).encode()
headers["Content-Type"] = "application/json"
req = urllib.request.Request(url, data=body, headers=headers, method=method)
with urllib.request.urlopen(req, timeout=URLOPEN_TIMEOUT_SECONDS) as response:
if response.status != HTTPStatus.NO_CONTENT:
return json.loads(response.read())
def get_recent_commit_count(pr_author, repo, token, since_days, max_count):
"""Return the number of recent commits by the author, up to max_count."""
if not pr_author:
return 0
params = {"author": pr_author, "per_page": max_count, "sha": "main"}
if since_days is not None:
params["since"] = (
datetime.now(timezone.utc) - timedelta(days=since_days)
).strftime("%Y-%m-%dT%H:%M:%SZ")
results = github_request(
"GET",
"/commits",
token,
repo,
params=params,
)
return len(results)
def github_results(path, repo, token, params=None):
"""Yield all results from a paginated GitHub list endpoint."""
params = dict(params or {})
page = 1
while True:
results = github_request(
"GET",
path,
token,
repo,
params={**params, "per_page": GITHUB_PER_PAGE, "page": page},
)
yield from results
if len(results) < GITHUB_PER_PAGE:
return
page += 1
def get_comment_ids_containing_marker(pr_number, repo, token, marker):
return [
comment["id"]
for comment in github_results(f"/issues/{pr_number}/comments", repo, token)
if marker in (comment["body"] or "")
and comment["user"]["login"] == "github-actions[bot]"
]
|
