Repository navigation
Expand file tree
/
Copy pathgithub_api.py
More file actions
130 lines (107 loc) · 4.29 KB
/
Copy pathgithub_api.py
File metadata and controls
130 lines (107 loc) · 4.29 KB
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
"""GitHub API helpers, plus the shared HTTP fetch primitive (`make_request`)
also used for npm/PyPI lookups. Kept out of update_versions.py so other
modules can import it without an import cycle.
"""
import json
import os
import re
import urllib.error
import urllib.request
from registry_utils import is_prerelease, normalize_release_version
def get_github_token() -> str | None:
"""Get GitHub token from environment."""
return os.environ.get("GITHUB_TOKEN")
def make_request(url: str, headers: dict | None = None) -> dict | list | str | None:
"""Make HTTP request and return JSON response."""
req_headers = {"User-Agent": "ACP-Registry-Version-Checker/1.0"}
if headers:
req_headers.update(headers)
# Add GitHub token if available and this is a GitHub API request
token = get_github_token()
if token and "api.github.com" in url:
req_headers["Authorization"] = f"token {token}"
try:
req = urllib.request.Request(url, headers=req_headers)
with urllib.request.urlopen(req, timeout=30) as response:
content = response.read().decode("utf-8")
try:
return json.loads(content)
except json.JSONDecodeError:
return content
except urllib.error.HTTPError as e:
if e.code == 404:
return None
if e.code >= 500:
return None
raise
except (urllib.error.URLError, TimeoutError, OSError):
return None
def is_github_repo(repo_url: str) -> bool:
return "github.com" in repo_url
def _github_owner_repo(repo_url: str) -> tuple[str, str] | None:
"""Extract (owner, repo) from a GitHub repository URL, stripping any `.git`."""
match = re.search(r"github\.com/([^/]+)/([^/]+)", repo_url)
if not match:
return None
owner, repo = match.groups()
if repo.endswith(".git"):
repo = repo[:-4]
return owner, repo
def _parse_release_digests(data: dict) -> dict[str, str]:
"""Extract {asset_name: hex_sha256} from a GitHub release payload."""
digests: dict[str, str] = {}
for a in data.get("assets", []):
if not isinstance(a, dict):
continue
name = a.get("name")
digest = a.get("digest", "")
if name and isinstance(digest, str) and digest.startswith("sha256:"):
digests[name] = digest.removeprefix("sha256:")
return digests
def get_github_latest_version(repo_url: str) -> str | None:
parsed = _github_owner_repo(repo_url)
if not parsed:
return None
owner, repo = parsed
data = make_request(f"https://api.github.com/repos/{owner}/{repo}/releases/latest")
if isinstance(data, dict):
tag = data.get("tag_name", "")
return normalize_release_version(tag.lstrip("v") if tag else None)
return None
def get_github_release_digests(repo_url: str, version: str) -> dict[str, str]:
"""Return {asset_filename: hex_sha256} for the release tagged `version`."""
parsed = _github_owner_repo(repo_url)
if not parsed:
return {}
owner, repo = parsed
# Try `v{version}` (common tag convention) then bare `{version}`.
for tag in (f"v{version}", version):
data = make_request(f"https://api.github.com/repos/{owner}/{repo}/releases/tags/{tag}")
if isinstance(data, dict):
return _parse_release_digests(data)
return {}
def get_github_release_versions(repo_url: str) -> set[str] | None:
"""Get stable GitHub release versions published for a repository."""
parsed = _github_owner_repo(repo_url)
if not parsed:
return None
owner, repo = parsed
api_url = f"https://api.github.com/repos/{owner}/{repo}/releases?per_page=100"
data = make_request(api_url)
if isinstance(data, list):
versions = set()
for release in data:
if not isinstance(release, dict):
continue
if release.get("draft") or release.get("prerelease"):
continue
tag = release.get("tag_name", "")
version = normalize_release_version(tag.lstrip("v") if tag else None)
if version and not is_prerelease(version):
versions.add(version)
if versions:
return versions
latest = get_github_latest_version(repo_url)
if latest:
return {latest}
return None