89 lines
3.2 KiB
Python
89 lines
3.2 KiB
Python
"""
|
|
Typed REST client for the Proxmox Image Conversion Backend API.
|
|
|
|
Connects to the backend at http://10.2.0.2:9000.
|
|
All API endpoints are under /api/v1/.
|
|
Derived from open-api.yaml — do not change models without updating the spec.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import requests
|
|
from urllib.parse import urljoin
|
|
from typing import Optional
|
|
|
|
BACKEND_URL = os.getenv("BACKEND_URL", "http://10.2.0.2:9000")
|
|
TIMEOUT = 30
|
|
ANALYZE_TIMEOUT = 300 # guestfish needs time to probe large disk images
|
|
|
|
|
|
class ApiError(Exception):
|
|
def __init__(self, status: int, detail: str):
|
|
self.status = status
|
|
self.detail = detail
|
|
super().__init__(f"HTTP {status}: {detail}")
|
|
|
|
|
|
class ApiClient:
|
|
"""Synchronous HTTP client for the backend API."""
|
|
|
|
def __init__(self, base_url: str = BACKEND_URL):
|
|
self.base_url = base_url.rstrip("/")
|
|
self._s = requests.Session()
|
|
self._s.headers.update({
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/json",
|
|
})
|
|
|
|
# ------------------------------------------------------------------
|
|
# Public API — matches open-api.yaml
|
|
# ------------------------------------------------------------------
|
|
|
|
def health(self) -> dict:
|
|
return self._get("/api/v1/health")
|
|
|
|
def analyze(self, vmid: int, filename: str, source_type: str = "upload") -> dict:
|
|
"""POST /api/v1/analyze — probe source file for OS, size, format."""
|
|
return self._post("/api/v1/analyze", {
|
|
"vmid": vmid,
|
|
"source_filename": filename,
|
|
"source_type": source_type,
|
|
}, timeout=ANALYZE_TIMEOUT)
|
|
|
|
def create_job(self, payload: dict) -> dict:
|
|
"""POST /api/v1/jobs — submit conversion job."""
|
|
return self._post("/api/v1/jobs", payload)
|
|
|
|
def get_job(self, job_id: str) -> dict:
|
|
"""GET /api/v1/jobs/{job_id} — poll job status."""
|
|
return self._get(f"/api/v1/jobs/{job_id}")
|
|
|
|
def cleanup_job(self, job_id: str, delete: bool) -> dict:
|
|
"""POST /api/v1/jobs/{job_id}/cleanup — remove or keep staging files."""
|
|
return self._post(f"/api/v1/jobs/{job_id}/cleanup", {"delete_staging_files": delete})
|
|
|
|
# ------------------------------------------------------------------
|
|
# Internal
|
|
# ------------------------------------------------------------------
|
|
|
|
def _get(self, path: str, timeout: int = TIMEOUT) -> dict:
|
|
return self._request("GET", path, timeout=timeout)
|
|
|
|
def _post(self, path: str, data: dict, timeout: int = TIMEOUT) -> dict:
|
|
return self._request("POST", path, data, timeout=timeout)
|
|
|
|
def _request(self, method: str, path: str, data: Optional[dict] = None, timeout: int = TIMEOUT) -> dict:
|
|
url = urljoin(self.base_url + "/", path.lstrip("/"))
|
|
try:
|
|
resp = self._s.request(method, url, json=data, timeout=timeout)
|
|
if resp.ok:
|
|
return resp.json()
|
|
detail = resp.text
|
|
try:
|
|
detail = resp.json().get("detail", detail)
|
|
except Exception:
|
|
pass
|
|
raise ApiError(resp.status_code, detail)
|
|
except requests.RequestException as exc:
|
|
raise ApiError(0, f"Connection failed: {exc}") from exc
|