98 lines
3.2 KiB
Python
98 lines
3.2 KiB
Python
"""Forgejo-backed registry client (contracts/registry-layout.md).
|
|
|
|
Listings live at ``listings/<listing_id>/{listing,solution}.json`` on the
|
|
registry repo's default branch, read via the Forgejo contents API. Solution
|
|
bundles attach as repo releases tagged ``listing.json.install_ref``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
|
|
|
|
class RegistryError(RuntimeError):
|
|
"""Raised when the registry returns an unexpected response."""
|
|
|
|
|
|
class RegistryClient:
|
|
def __init__(
|
|
self,
|
|
base_url: str,
|
|
token: str,
|
|
repo: str = "m2/market-registry",
|
|
auth_user: str = "m2",
|
|
) -> None:
|
|
owner, _, name = repo.partition("/")
|
|
self._owner = owner
|
|
self._repo = name
|
|
self._client = httpx.Client(
|
|
base_url=base_url.rstrip("/"),
|
|
auth=(auth_user, token),
|
|
)
|
|
|
|
def __enter__(self) -> "RegistryClient":
|
|
return self
|
|
|
|
def __exit__(self, *exc_info: object) -> None:
|
|
self.close()
|
|
|
|
def close(self) -> None:
|
|
self._client.close()
|
|
|
|
def _repo_api(self, path: str) -> str:
|
|
return f"/api/v1/repos/{self._owner}/{self._repo}{path}"
|
|
|
|
def _get_json_contents(self, repo_path: str) -> dict:
|
|
resp = self._client.get(self._repo_api(f"/contents/{repo_path}"))
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
content = base64.b64decode(data["content"])
|
|
return json.loads(content)
|
|
|
|
def get_listing(self, listing_id: str) -> dict:
|
|
return self._get_json_contents(f"listings/{listing_id}/listing.json")
|
|
|
|
def get_solution(self, listing_id: str) -> dict:
|
|
return self._get_json_contents(f"listings/{listing_id}/solution.json")
|
|
|
|
def download_bundle(self, install_ref: str, dest_dir: str | Path) -> Path:
|
|
resp = self._client.get(self._repo_api(f"/releases/tags/{install_ref}"))
|
|
resp.raise_for_status()
|
|
release = resp.json()
|
|
assets = release.get("assets") or []
|
|
bundle = next((a for a in assets if a["name"].endswith(".tar.gz")), None)
|
|
if bundle is None:
|
|
raise RegistryError(
|
|
f"release {install_ref!r} has no .tar.gz bundle asset"
|
|
)
|
|
|
|
dest_dir = Path(dest_dir)
|
|
dest_dir.mkdir(parents=True, exist_ok=True)
|
|
dest_path = dest_dir / bundle["name"]
|
|
download_url = bundle.get("browser_download_url") or bundle["url"]
|
|
|
|
with self._client.stream("GET", download_url) as stream:
|
|
stream.raise_for_status()
|
|
with dest_path.open("wb") as fh:
|
|
for chunk in stream.iter_bytes():
|
|
fh.write(chunk)
|
|
|
|
return dest_path
|
|
|
|
@staticmethod
|
|
def verify_bundle(path: str | Path, content_hash: str) -> bool:
|
|
algo, _, expected = content_hash.partition(":")
|
|
if algo != "sha256" or not expected:
|
|
raise ValueError(f"unsupported content_hash format: {content_hash!r}")
|
|
|
|
digest = hashlib.sha256()
|
|
with Path(path).open("rb") as fh:
|
|
for chunk in iter(lambda: fh.read(65536), b""):
|
|
digest.update(chunk)
|
|
|
|
return digest.hexdigest() == expected
|