install the publication mechanism

.publication/lib/giteaapi.py

@@ -0,0 +1,69 @@
+"""Minimal read-only Gitea Actions client.
+
+The token is read from the environment by name and never logged, echoed, or
+placed in a URL. Only GET is implemented: this client exists to prove state,
+never to change it.
+"""
+
+import json
+import os
+import urllib.error
+import urllib.parse
+import urllib.request
+
+DEFAULT_PAGE_SIZE = 50
+MAX_PAGES = 40
+
+
+class ApiError(Exception):
+    pass
+
+
+class GiteaClient:
+    def __init__(self, base_url, token_env="GITEA_READ_TOKEN", opener=None, timeout=20):
+        self.base_url = base_url.rstrip("/")
+        self._token = os.environ.get(token_env)
+        if not self._token:
+            raise ApiError(f"no read token in ${token_env}")
+        self._opener = opener or urllib.request.build_opener()
+        self._timeout = timeout
+
+    def _get(self, path, params=None):
+        url = f"{self.base_url}/api/v1{path}"
+        if params:
+            url += "?" + urllib.parse.urlencode(params)
+        req = urllib.request.Request(url, method="GET")
+        req.add_header("Authorization", f"token {self._token}")
+        req.add_header("Accept", "application/json")
+        try:
+            with self._opener.open(req, timeout=self._timeout) as resp:
+                return json.loads(resp.read().decode("utf-8"))
+        except urllib.error.HTTPError as exc:
+            raise ApiError(f"GET {path} -> HTTP {exc.code}")
+        except urllib.error.URLError as exc:
+            raise ApiError(f"GET {path} -> {exc.reason}")
+
+    def _paged(self, path, params, key):
+        out, page = [], 1
+        while page <= MAX_PAGES:
+            q = dict(params or {})
+            q.update({"page": page, "limit": DEFAULT_PAGE_SIZE})
+            body = self._get(path, q)
+            batch = body.get(key, []) if isinstance(body, dict) else body
+            if not batch:
+                break
+            out.extend(batch)
+            if len(batch) < DEFAULT_PAGE_SIZE:
+                break
+            page += 1
+        else:
+            raise ApiError(f"GET {path} exceeded {MAX_PAGES} pages; refusing to guess")
+        return out
+
+    def list_runs(self, repository, head_sha):
+        return self._paged(f"/repos/{repository}/actions/runs",
+                           {"head_sha": head_sha}, "workflow_runs")
+
+    def list_jobs(self, repository, run_id):
+        return self._paged(f"/repos/{repository}/actions/runs/{run_id}/jobs",
+                           None, "jobs")