"""SourceCraft REST API client (ARCHITECTURE ยง2.4). The API drops TLS connections now and then: every call retries. Runs don't carry a commit sha, so the uploads CI prints it from its first cube (`check/meta`, ADR-014); `run_commit_sha` reads it back. """ import asyncio import logging import re from collections.abc import Awaitable, Callable from types import TracebackType from typing import Any, Self import httpx from app.config import SourceCraftSettings from app.sourcecraft.models import CubeLog, Run log = logging.getLogger(__name__) META_TASK = "check" META_CUBE = "meta" _SHA_LINE = re.compile(r"^TGIT_COMMIT_SHA=([0-9a-f]{40})\s*$", re.MULTILINE) Sleep = Callable[[float], Awaitable[None]] class SourceCraftError(Exception): def __init__(self, status: int, code: str, message: str) -> None: super().__init__(f"{status} {code}: {message}") self.status = status self.code = code self.message = message class SourceCraftClient: def __init__( self, settings: SourceCraftSettings, transport: httpx.AsyncBaseTransport | None = None, sleep: Sleep = asyncio.sleep, ) -> None: self._settings = settings self._sleep = sleep self._http = httpx.AsyncClient( base_url=settings.api_url, headers={"Authorization": f"Bearer {settings.token.get_secret_value()}"}, timeout=settings.timeout, transport=transport, ) async def __aenter__(self) -> Self: return self async def __aexit__( self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: TracebackType | None ) -> None: await self.aclose() async def aclose(self) -> None: await self._http.aclose() def run_url(self, repo: str, run: str) -> str: return self._settings.run_url_template.format(repo=repo, run=run) async def list_runs(self, repo: str, page_size: int = 20) -> list[Run]: """Newest first. One page is enough for the watcher: it only looks for runs of recent pushes.""" data = await self._get(f"/repos/{repo}/cicd/runs", params={"page_size": page_size}) return [Run.model_validate(r) for r in data.get("runs", [])] async def get_run(self, repo: str, run: str) -> Run: return Run.model_validate(await self._get(f"/repos/{repo}/cicd/runs/{run}")) async def get_cube_logs(self, repo: str, run: str, workflow: str, task: str, cube: str) -> CubeLog: """Concatenate log pages in order; a page that isn't complete yet is the last readable one.""" parts: list[str] = [] done = False page = 1 while True: try: data = await self._get(f"/repos/{repo}/cicd/logs/{run}/{workflow}/{task}/{cube}", params={"page": page}) except SourceCraftError as e: if e.status == 404 and page > 1: break raise parts.append(data.get("logs", "")) done = bool(data.get("done", False)) if not data.get("page_complete", False): break page += 1 return CubeLog(text="".join(parts), done=done) async def run_commit_sha(self, repo: str, run: Run) -> str | None: """The commit a run was started for, read from the meta cube's log (None if not printed yet).""" for workflow in run.workflows: if run.task(workflow.slug, META_TASK) is None: continue try: logs = await self.get_cube_logs(repo, run.slug, workflow.slug, META_TASK, META_CUBE) except SourceCraftError as e: if e.status == 404: return None raise match = _SHA_LINE.search(logs.text) return match.group(1) if match else None return None async def _get(self, path: str, params: dict[str, Any] | None = None) -> dict[str, Any]: attempts = self._settings.max_attempts for attempt in range(1, attempts + 1): try: response = await self._http.get(path, params=params) except httpx.TransportError as e: if attempt == attempts: raise log.warning("sourcecraft %s: %s, retry %d/%d", path, type(e).__name__, attempt, attempts - 1) else: if response.status_code < 500: return self._parse(response) if attempt == attempts: return self._parse(response) log.warning("sourcecraft %s: HTTP %d, retry %d/%d", path, response.status_code, attempt, attempts - 1) await self._sleep(min(0.5 * 2 ** (attempt - 1), 8.0)) raise AssertionError("unreachable") @staticmethod def _parse(response: httpx.Response) -> dict[str, Any]: try: body = response.json() except ValueError: body = {} if response.status_code >= 400: raise SourceCraftError( response.status_code, str(body.get("error_code", "")), str(body.get("message", response.text[:200])), ) if not isinstance(body, dict): raise SourceCraftError(response.status_code, "BadResponse", "expected a JSON object") return body