nessi
4554b00b73
Add SEED_DEMO_DATA environment variable to control demo data population, enhance setup wizard with welcome screen and theme toggle, add smooth animations for wizard transitions, improve dashboard endpoint to return structured objects for last_syncs and faulty_nodes instead of raw models, implement comprehensive error handling in cluster sync with failed status tracking and audit logging, fix Proxmox provider
65 lines
2.9 KiB
Python
65 lines
2.9 KiB
Python
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from app.services.providers.base import Provider, ProviderConnection
|
|
|
|
|
|
class ProxmoxProvider(Provider):
|
|
name = "proxmox"
|
|
|
|
def auth_header(self, token: str) -> str:
|
|
token = token.strip()
|
|
if token.startswith("PVEAPIToken="):
|
|
return token
|
|
return f"PVEAPIToken={token}"
|
|
|
|
async def test_connection(self, connection: ProviderConnection) -> dict[str, Any]:
|
|
headers = {"Authorization": self.auth_header(connection.token)}
|
|
async with httpx.AsyncClient(verify=connection.verify_tls, timeout=10) as client:
|
|
response = await client.get(
|
|
f"{connection.api_url.rstrip('/')}/api2/json/version",
|
|
headers=headers,
|
|
)
|
|
response.raise_for_status()
|
|
return response.json().get("data", {})
|
|
|
|
async def sync_inventory(self, connection: ProviderConnection) -> dict[str, list[dict[str, Any]]]:
|
|
headers = {"Authorization": self.auth_header(connection.token)}
|
|
async with httpx.AsyncClient(verify=connection.verify_tls, timeout=20) as client:
|
|
resources = await client.get(
|
|
f"{connection.api_url.rstrip('/')}/api2/json/cluster/resources",
|
|
headers=headers,
|
|
)
|
|
resources.raise_for_status()
|
|
data = resources.json().get("data", [])
|
|
|
|
nodes = [item for item in data if item.get("type") == "node"]
|
|
workloads = [item for item in data if item.get("type") in {"qemu", "lxc"}]
|
|
networks = await self.list_networks(connection)
|
|
return {"nodes": nodes, "workloads": workloads, "networks": networks}
|
|
|
|
async def list_networks(self, connection: ProviderConnection) -> list[dict[str, Any]]:
|
|
headers = {"Authorization": self.auth_header(connection.token)}
|
|
async with httpx.AsyncClient(verify=connection.verify_tls, timeout=20) as client:
|
|
resources = await client.get(
|
|
f"{connection.api_url.rstrip('/')}/api2/json/cluster/resources",
|
|
headers=headers,
|
|
)
|
|
resources.raise_for_status()
|
|
data = resources.json().get("data", [])
|
|
return [item for item in data if item.get("type") in {"network", "sdn"}]
|
|
|
|
async def preview_rules(self, connection: ProviderConnection, rules: list[dict[str, Any]]) -> dict[str, Any]:
|
|
return {
|
|
"provider": self.name,
|
|
"read_only": connection.read_only,
|
|
"generated": rules,
|
|
"warnings": ["Preview only. No Proxmox firewall changes were sent."],
|
|
}
|
|
|
|
async def apply_rules(self, connection: ProviderConnection, rules: list[dict[str, Any]]) -> dict[str, Any]:
|
|
if connection.read_only:
|
|
return {"applied": False, "reason": "Cluster is read-only", "rules": rules}
|
|
return {"applied": False, "reason": "Apply adapter intentionally requires explicit implementation", "rules": rules}
|