接入 OpenSandbox Kata Runner 控制面

src/gitea_dynamic_runner/opensandbox.py

@@ -0,0 +1,79 @@
+"""Minimal client for the OpenSandbox lifecycle API."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from urllib.parse import quote
+
+from aiohttp import ClientResponseError, ClientSession
+
+
+class OpenSandboxClient:
+    def __init__(
+        self,
+        *,
+        api_url: str,
+        api_key_file: Path | None = None,
+    ) -> None:
+        self.api_url = api_url.rstrip("/")
+        self.api_key_file = api_key_file
+        self.session: ClientSession | None = None
+
+    async def __aenter__(self) -> OpenSandboxClient:
+        headers = {}
+        if self.api_key_file is not None:
+            headers["OPEN-SANDBOX-API-KEY"] = self.api_key_file.read_text().strip()
+        self.session = ClientSession(headers=headers, raise_for_status=True)
+        return self
+
+    async def __aexit__(self, *_: object) -> None:
+        if self.session is not None:
+            await self.session.close()
+
+    async def create(
+        self,
+        *,
+        pool: str,
+        timeout: int,
+        entrypoint: list[str],
+        env: dict[str, str],
+        metadata: dict[str, str],
+    ) -> dict[str, object]:
+        response = await self._request(
+            "POST",
+            "/v1/sandboxes",
+            json={
+                "timeout": timeout,
+                "entrypoint": entrypoint,
+                "env": env,
+                "metadata": metadata,
+                "extensions": {"poolRef": pool},
+            },
+        )
+        return await response.json()
+
+    async def get(self, sandbox_id: str) -> dict[str, object] | None:
+        try:
+            response = await self._request(
+                "GET", f"/v1/sandboxes/{quote(sandbox_id, safe='')}"
+            )
+        except ClientResponseError as error:
+            if error.status == 404:
+                return None
+            raise
+        return await response.json()
+
+    async def delete(self, sandbox_id: str) -> None:
+        try:
+            response = await self._request(
+                "DELETE", f"/v1/sandboxes/{quote(sandbox_id, safe='')}"
+            )
+            response.release()
+        except ClientResponseError as error:
+            if error.status != 404:
+                raise
+
+    async def _request(self, method: str, path: str, **kwargs: object):
+        if self.session is None:
+            raise RuntimeError("OpenSandboxClient is not open")
+        return await self.session.request(method, f"{self.api_url}{path}", **kwargs)