80 lines
2.4 KiB
Python
80 lines
2.4 KiB
Python
"""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)
|