Add openstack_client.py

openstack_client.py

@@ -0,0 +1,91 @@
+"""Inventory client: per-resource detail from Nova, Cinder, Neutron, Keystone.
+
+Endpoints come from the Keystone token catalog (public interface), so this
+works both inside the cluster and externally (console.connectvm.com).
+"""
+import httpx
+
+from cloudkitty_client import KeystoneAuth
+from config import Settings
+
+NOVA_MICROVERSION = "2.88"  # embedded flavor objects in server details
+
+
+class OpenStackClient:
+    def __init__(self, settings: Settings):
+        self._s = settings
+        self._auth = KeystoneAuth(settings)
+        self._client = httpx.AsyncClient(
+            verify=settings.verify_ssl, timeout=settings.http_timeout
+        )
+        self._catalog: list | None = None
+        self._project_id: str | None = None
+
+    async def close(self):
+        await self._client.aclose()
+
+    async def _ensure_catalog(self):
+        if self._catalog is None:
+            body = await self._auth.token_body()
+            token = body.get("token", {})
+            self._catalog = token.get("catalog", [])
+            self._project_id = token.get("project", {}).get("id")
+
+    def _endpoint(self, service_type: str, override: str = "") -> str:
+        if override:
+            return override.rstrip("/")
+        for svc in self._catalog or []:
+            if svc.get("type") == service_type:
+                for ep in svc.get("endpoints", []):
+                    if ep.get("interface") == "public":
+                        return ep["url"].rstrip("/")
+        raise RuntimeError(f"no public endpoint for service type '{service_type}'")
+
+    async def _get(self, url: str, params: dict | None = None, headers: dict | None = None):
+        token = await self._auth.token()
+        h = {"X-Auth-Token": token}
+        h.update(headers or {})
+        r = await self._client.get(url, params=params, headers=h)
+        if r.status_code == 401:
+            h["X-Auth-Token"] = await self._auth.token(force=True)
+            r = await self._client.get(url, params=params, headers=h)
+        r.raise_for_status()
+        return r.json()
+
+    async def projects(self) -> dict[str, str]:
+        """Map of project name -> project id."""
+        data = await self._get(f"{self._s.keystone_auth_url}/v3/projects")
+        return {p["name"]: p["id"] for p in data.get("projects", [])}
+
+    async def servers(self, project_id: str) -> list[dict]:
+        await self._ensure_catalog()
+        base = self._endpoint("compute", self._s.nova_url)
+        data = await self._get(
+            f"{base}/servers/detail",
+            params={"all_tenants": 1, "project_id": project_id, "limit": 1000},
+            headers={"X-OpenStack-Nova-API-Version": NOVA_MICROVERSION},
+        )
+        return data.get("servers", [])
+
+    async def volumes(self, project_id: str) -> list[dict]:
+        await self._ensure_catalog()
+        base = self._endpoint("volumev3", self._s.cinder_url).replace(
+            "%(project_id)s", self._project_id or ""
+        )
+        data = await self._get(
+            f"{base}/volumes/detail",
+            params={"all_tenants": 1, "limit": 1000},
+        )
+        return [
+            v
+            for v in data.get("volumes", [])
+            if v.get("os-vol-tenant-attr:tenant_id") == project_id
+        ]
+
+    async def floatingips(self, project_id: str) -> list[dict]:
+        await self._ensure_catalog()
+        base = self._endpoint("network", self._s.neutron_url)
+        data = await self._get(
+            f"{base}/v2.0/floatingips", params={"tenant_id": project_id}
+        )
+        return data.get("floatingips", [])