157 lines
4.6 KiB
Python
157 lines
4.6 KiB
Python
"""Thin Python SDK client for the ``/v1`` REST API.
|
|
|
|
Capability: ``platform-sdk``.
|
|
|
|
Uses ``httpx`` directly so the client can talk to any deployed cloud-runtime
|
|
process over HTTP, or to a FastAPI ``TestClient`` instance in tests.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
|
|
class CloudAPIError(httpx.HTTPStatusError):
|
|
def __init__(self, response: httpx.Response, detail: str) -> None:
|
|
super().__init__(
|
|
f"cloud API request failed with status {response.status_code}: {detail}",
|
|
request=response.request,
|
|
response=response,
|
|
)
|
|
self.status_code = response.status_code
|
|
self.detail = detail
|
|
|
|
|
|
class CloudAuthorizationError(CloudAPIError):
|
|
pass
|
|
|
|
|
|
class CloudClient:
|
|
"""A minimal Python wrapper for the platform SDK's ``/v1`` routes."""
|
|
|
|
def __init__(
|
|
self,
|
|
base_url: str,
|
|
*,
|
|
http_client: httpx.Client | Any = None,
|
|
api_prefix: str = "/v1",
|
|
token: str | None = None,
|
|
auth: httpx.Auth | None = None,
|
|
) -> None:
|
|
if token is not None and auth is not None:
|
|
raise ValueError("token and auth are mutually exclusive")
|
|
self._base_url = base_url.rstrip("/")
|
|
self._api_prefix = api_prefix.rstrip("/")
|
|
self._headers = {"Authorization": f"Bearer {token}"} if token else None
|
|
self._auth = auth
|
|
if http_client is None:
|
|
self._http = httpx.Client(base_url=self._base_url)
|
|
self._owns_client = True
|
|
else:
|
|
self._http = http_client
|
|
self._owns_client = False
|
|
|
|
def close(self) -> None:
|
|
if self._owns_client:
|
|
self._http.close()
|
|
|
|
def __enter__(self) -> "CloudClient":
|
|
return self
|
|
|
|
def __exit__(self, *exc: object) -> None:
|
|
self.close()
|
|
|
|
# ------------------------------------------------------------------- tasks
|
|
|
|
def submit_task(
|
|
self,
|
|
*,
|
|
goal: str | None = None,
|
|
workflow_definition_id: str | None = None,
|
|
driver_type: str | None = None,
|
|
capability_tags: list[str] | None = None,
|
|
) -> dict[str, Any]:
|
|
payload: dict[str, Any] = {
|
|
"goal": goal,
|
|
"workflow_definition_id": workflow_definition_id,
|
|
}
|
|
if driver_type is not None or capability_tags is not None:
|
|
payload["constraints"] = {
|
|
"driver_type": driver_type,
|
|
"capability_tags": list(capability_tags or []),
|
|
}
|
|
resp = self._request("POST", "/tasks", json=payload)
|
|
return resp.json()
|
|
|
|
def get_task_status(self, task_id: str) -> dict[str, Any]:
|
|
resp = self._request("GET", f"/tasks/{task_id}")
|
|
return resp.json()
|
|
|
|
# ----------------------------------------------------------------- devices
|
|
|
|
def list_devices(self) -> list[dict[str, Any]]:
|
|
resp = self._request("GET", "/devices")
|
|
return resp.json()
|
|
|
|
def list_hosts(self) -> list[dict[str, Any]]:
|
|
resp = self._request("GET", "/hosts")
|
|
return resp.json()
|
|
|
|
# ----------------------------------------------------------------- plugins
|
|
|
|
def list_plugins(self) -> list[dict[str, Any]]:
|
|
resp = self._request("GET", "/plugins")
|
|
return resp.json()
|
|
|
|
def register_plugin(
|
|
self,
|
|
*,
|
|
name: str,
|
|
version: str,
|
|
entry_point_kind: str,
|
|
target: str,
|
|
) -> dict[str, Any]:
|
|
payload = {
|
|
"name": name,
|
|
"version": version,
|
|
"entry_point_kind": entry_point_kind,
|
|
"target": target,
|
|
}
|
|
resp = self._request("POST", "/plugins", json=payload)
|
|
return resp.json()
|
|
|
|
# ------------------------------------------------------------------ helpers
|
|
|
|
def _url(self, path: str) -> str:
|
|
return f"{self._base_url}{self._api_prefix}{path}"
|
|
|
|
def _request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
json: dict[str, Any] | None = None,
|
|
) -> httpx.Response:
|
|
response = self._http.request(
|
|
method,
|
|
self._url(path),
|
|
json=json,
|
|
headers=self._headers,
|
|
auth=self._auth,
|
|
)
|
|
if response.is_success:
|
|
return response
|
|
try:
|
|
payload = response.json()
|
|
except ValueError:
|
|
payload = {}
|
|
detail = payload.get("detail") or "request rejected"
|
|
error_type = (
|
|
CloudAuthorizationError
|
|
if response.status_code in {401, 403}
|
|
else CloudAPIError
|
|
)
|
|
raise error_type(response, str(detail))
|