Add cloudkitty_client.py

cloudkitty_client.py

@@ -0,0 +1,103 @@
+"""Async client for Keystone auth + CloudKitty v2 summary API."""
+import time
+
+import httpx
+
+from config import Settings
+
+
+class KeystoneAuth:
+    def __init__(self, settings: Settings):
+        self._s = settings
+        self._token: str | None = None
+        self._body: dict | None = None
+        self._expires_at: float = 0.0
+
+    async def token(self, force: bool = False) -> str:
+        await self.token_body(force=force)
+        return self._token
+
+    async def token_body(self, force: bool = False) -> dict:
+        """Full Keystone token response (catalog, project, expires)."""
+        if not force and self._body and time.time() < self._expires_at:
+            return self._body
+
+        payload = {
+            "auth": {
+                "identity": {
+                    "methods": ["password"],
+                    "password": {
+                        "user": {
+                            "name": self._s.os_username,
+                            "domain": {"name": self._s.os_user_domain},
+                            "password": self._s.os_password,
+                        }
+                    },
+                },
+                "scope": {
+                    "project": {
+                        "name": self._s.os_project_name,
+                        "domain": {"name": self._s.os_project_domain},
+                    }
+                },
+            }
+        }
+        async with httpx.AsyncClient(
+            verify=self._s.verify_ssl, timeout=self._s.http_timeout
+        ) as client:
+            r = await client.post(
+                f"{self._s.keystone_auth_url}/v3/auth/tokens", json=payload
+            )
+            r.raise_for_status()
+        self._token = r.headers["X-Subject-Token"]
+        self._body = r.json()
+        self._expires_at = time.time() + self._s.token_ttl_seconds
+        return self._body
+
+
+class CloudKittyClient:
+    def __init__(self, settings: Settings):
+        self._auth = KeystoneAuth(settings)
+        self._client = httpx.AsyncClient(
+            base_url=settings.cloudkitty_url,
+            verify=settings.verify_ssl,
+            timeout=settings.http_timeout,
+        )
+
+    async def close(self):
+        await self._client.aclose()
+
+    async def _request(self, method: str, path: str, **kwargs) -> httpx.Response:
+        token = await self._auth.token()
+        r = await self._client.request(
+            method, path, headers={"X-Auth-Token": token}, **kwargs
+        )
+        if r.status_code == 401:
+            # Token rejected despite cache — refresh once and retry
+            token = await self._auth.token(force=True)
+            r = await self._client.request(
+                method, path, headers={"X-Auth-Token": token}, **kwargs
+            )
+        r.raise_for_status()
+        return r
+
+    async def summary(
+        self,
+        begin=None,
+        end=None,
+        groupby: tuple[str, ...] = ("tenant",),
+        extra_params: dict | None = None,
+    ) -> list[dict]:
+        """Return summary rows as list of dicts keyed by CloudKitty columns."""
+        params: dict = {"groupby": ",".join(groupby)}
+        if begin is not None:
+            params["begin"] = begin.isoformat()
+        if end is not None:
+            params["end"] = end.isoformat()
+        if extra_params:
+            params.update(extra_params)
+
+        r = await self._request("GET", "/v2/summary", params=params)
+        data = r.json()
+        cols = data.get("columns", [])
+        return [dict(zip(cols, row)) for row in data.get("results", [])]