merge: cli-registry-client (US1 wave 1)

cli/src/m2_market/registry.py

@@ -0,0 +1,98 @@
+"""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