Claus Lohmar
7eb2d6bb45
- New GET /api/v1/storage/pools returns available lvmthin/zfspool/rbd/dir pools - Analysis result template now shows dropdown when >1 pool available - Falls back to text input with single pool or on discovery failure
107 lines
3.9 KiB
Python
107 lines
3.9 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
|
|
|
|
|
|
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 — start async analysis, returns analysis_id."""
|
|
return self._post("/api/v1/analyze", {
|
|
"vmid": vmid,
|
|
"source_filename": filename,
|
|
"source_type": source_type,
|
|
})
|
|
|
|
def get_analysis(self, analysis_id: str) -> dict:
|
|
"""GET /api/v1/analyze/{analysis_id} — poll analysis status."""
|
|
return self._get(f"/api/v1/analyze/{analysis_id}")
|
|
|
|
def get_storage_pools(self) -> dict:
|
|
"""GET /api/v1/storage/pools — list available Proxmox storage pools."""
|
|
return self._get("/api/v1/storage/pools")
|
|
|
|
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, session_id: str = "") -> 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,
|
|
"session_id": session_id,
|
|
})
|
|
|
|
def clone_vm(self, source_vmid: int, target_vmid: int, target_name: str) -> dict:
|
|
"""POST /api/v1/clone — clone an existing VM."""
|
|
return self._post("/api/v1/clone", {
|
|
"source_vmid": source_vmid,
|
|
"target_vmid": target_vmid,
|
|
"target_name": target_name,
|
|
})
|
|
|
|
# ------------------------------------------------------------------
|
|
# 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
|