buildbot-nix/buildbot_nix/github_projects.py

193 lines
5.2 KiB
Python
Raw Normal View History

2023-12-26 20:56:36 +00:00
import contextlib
2023-09-10 08:11:56 +00:00
import http.client
import json
import urllib.request
from pathlib import Path
from tempfile import NamedTemporaryFile
2023-09-10 08:11:56 +00:00
from typing import Any
from twisted.python import log
2023-12-26 20:56:36 +00:00
class GithubError(Exception):
pass
2023-09-10 08:11:56 +00:00
class HttpResponse:
def __init__(self, raw: http.client.HTTPResponse) -> None:
self.raw = raw
def json(self) -> Any:
return json.load(self.raw)
def headers(self) -> http.client.HTTPMessage:
return self.raw.headers
def http_request(
url: str,
method: str = "GET",
2023-12-26 20:56:36 +00:00
headers: dict[str, str] | None = None,
2023-09-10 08:11:56 +00:00
data: dict[str, Any] | None = None,
) -> HttpResponse:
body = None
if data:
body = json.dumps(data).encode("ascii")
2023-12-26 20:56:36 +00:00
if headers is None:
headers = {}
2023-09-10 08:11:56 +00:00
headers = headers.copy()
headers["User-Agent"] = "buildbot-nix"
2023-12-26 20:56:36 +00:00
if not url.startswith("https:"):
msg = "url must be https: {url}"
raise GithubError(msg)
req = urllib.request.Request( # noqa: S310
url, headers=headers, method=method, data=body
)
2023-10-12 13:59:26 +00:00
try:
2023-12-26 20:56:36 +00:00
resp = urllib.request.urlopen(req) # noqa: S310
2023-10-12 13:59:26 +00:00
except urllib.request.HTTPError as e:
resp_body = ""
2023-12-26 20:56:36 +00:00
with contextlib.suppress(OSError, UnicodeDecodeError):
resp_body = e.fp.read().decode("utf-8", "replace")
2023-12-26 20:56:36 +00:00
msg = f"Request for {method} {url} failed with {e.code} {e.reason}: {resp_body}"
raise GithubError(msg) from e
2023-09-10 08:11:56 +00:00
return HttpResponse(resp)
def paginated_github_request(url: str, token: str) -> list[dict[str, Any]]:
next_url: str | None = url
2023-10-12 13:59:26 +00:00
items = []
2023-09-10 08:11:56 +00:00
while next_url:
try:
res = http_request(
next_url,
2023-10-12 13:59:26 +00:00
headers={"Authorization": f"Bearer {token}"},
2023-09-10 08:11:56 +00:00
)
except OSError as e:
2023-12-26 20:56:36 +00:00
msg = f"failed to fetch {next_url}: {e}"
raise GithubError(msg) from e
2023-09-10 08:11:56 +00:00
next_url = None
link = res.headers()["Link"]
if link is not None:
links = link.split(", ")
for link in links: # pagination
link_parts = link.split(";")
if link_parts[1].strip() == 'rel="next"':
next_url = link_parts[0][1:-1]
2023-10-12 13:59:26 +00:00
items += res.json()
return items
2023-09-10 08:11:56 +00:00
def slugify_project_name(name: str) -> str:
return name.replace(".", "-").replace("/", "-")
2023-09-10 08:11:56 +00:00
class GithubProject:
2023-10-12 13:59:26 +00:00
def __init__(self, data: dict[str, Any]) -> None:
self.data = data
@property
def repo(self) -> str:
return self.data["name"]
@property
def owner(self) -> str:
return self.data["owner"]["login"]
2023-09-10 08:11:56 +00:00
@property
def name(self) -> str:
2023-10-12 13:59:26 +00:00
return self.data["full_name"]
2023-09-10 08:11:56 +00:00
@property
def url(self) -> str:
2023-10-12 13:59:26 +00:00
return self.data["html_url"]
2023-09-10 08:11:56 +00:00
@property
2023-12-26 20:56:36 +00:00
def project_id(self) -> str:
return slugify_project_name(self.data["full_name"])
2023-09-10 08:11:56 +00:00
@property
def default_branch(self) -> str:
2023-10-12 13:59:26 +00:00
return self.data["default_branch"]
2023-09-10 08:11:56 +00:00
@property
def topics(self) -> list[str]:
2023-10-12 13:59:26 +00:00
return self.data["topics"]
@property
def belongs_to_org(self) -> bool:
return self.data["owner"]["type"] == "Organization"
2023-10-12 13:59:26 +00:00
def create_project_hook(
2023-12-26 20:56:36 +00:00
owner: str,
repo: str,
token: str,
webhook_url: str,
webhook_secret: str,
) -> None:
2023-10-12 13:59:26 +00:00
hooks = paginated_github_request(
2023-12-26 20:56:36 +00:00
f"https://api.github.com/repos/{owner}/{repo}/hooks?per_page=100",
token,
2023-10-12 13:59:26 +00:00
)
config = dict(
2023-12-26 20:56:36 +00:00
url=webhook_url,
content_type="json",
insecure_ssl="0",
secret=webhook_secret,
)
2023-10-12 13:59:26 +00:00
data = dict(name="web", active=True, events=["push", "pull_request"], config=config)
headers = {
"Authorization": f"Bearer {token}",
"Accept": "application/vnd.github+json",
"Content-Type": "application/json",
"X-GitHub-Api-Version": "2022-11-28",
}
for hook in hooks:
if hook["config"]["url"] == webhook_url:
log.msg(f"hook for {owner}/{repo} already exists")
return
http_request(
f"https://api.github.com/repos/{owner}/{repo}/hooks",
method="POST",
headers=headers,
data=data,
)
2023-09-10 08:11:56 +00:00
def refresh_projects(github_token: str, repo_cache_file: Path) -> None:
repos = []
for repo in paginated_github_request(
2023-10-27 08:49:40 +00:00
"https://api.github.com/user/repos?per_page=100",
github_token,
):
if not repo["permissions"]["admin"]:
2023-10-27 08:49:40 +00:00
name = repo["full_name"]
log.msg(
2023-12-26 20:56:36 +00:00
f"skipping {name} because we do not have admin privileges, needed for hook management",
2023-10-27 08:49:40 +00:00
)
else:
repos.append(repo)
with NamedTemporaryFile("w", delete=False, dir=repo_cache_file.parent) as f:
2023-12-26 20:56:36 +00:00
path = Path(f.name)
try:
f.write(json.dumps(repos))
f.flush()
2023-12-26 20:56:36 +00:00
path.rename(repo_cache_file)
except OSError:
2023-12-26 20:56:36 +00:00
path.unlink()
raise
2023-09-10 08:11:56 +00:00
def load_projects(github_token: str, repo_cache_file: Path) -> list[GithubProject]:
2023-10-26 08:17:49 +00:00
if not repo_cache_file.exists():
log.msg("fetching github repositories")
refresh_projects(github_token, repo_cache_file)
2023-10-26 08:17:49 +00:00
repos: list[dict[str, Any]] = json.loads(repo_cache_file.read_text())
2023-09-10 08:11:56 +00:00
return [GithubProject(repo) for repo in repos]