Extract shared NPM API client

dns_sync/npm_client.py

@@ -0,0 +1,127 @@
+import os
+from typing import Any
+
+import requests
+import urllib3
+
+
+class NpmClient:
+    """Authenticated client for the Nginx Proxy Manager API."""
+
+    def __init__(self, config):
+        self.config = config
+
+        self.identity = os.getenv("NPM_IDENTITY")
+        self.secret = os.getenv("NPM_SECRET")
+
+        if not self.identity:
+            raise RuntimeError("NPM_IDENTITY is required")
+
+        if not self.secret:
+            raise RuntimeError("NPM_SECRET is required")
+
+        self.session = requests.Session()
+        self.token: str | None = None
+
+        if not config.proxy_verify_tls:
+            urllib3.disable_warnings(
+                urllib3.exceptions.InsecureRequestWarning
+            )
+
+    @property
+    def api_url(self) -> str:
+        return f"{self.config.proxy_url}/api"
+
+    def login(self) -> None:
+        response = self.session.post(
+            f"{self.api_url}/tokens",
+            json={
+                "identity": self.identity,
+                "secret": self.secret,
+            },
+            timeout=15,
+            verify=self.config.proxy_verify_tls,
+        )
+
+        if response.status_code != 200:
+            raise RuntimeError(
+                "NPM login failed: "
+                f"HTTP {response.status_code}: "
+                f"{response.text[:300]}"
+            )
+
+        data = response.json()
+        token = data.get("token")
+
+        if not token:
+            raise RuntimeError(
+                "NPM login succeeded but no token was returned"
+            )
+
+        self.token = str(token)
+
+        self.session.headers.update(
+            {
+                "Authorization": f"Bearer {self.token}",
+                "Accept": "application/json",
+            }
+        )
+
+    def _ensure_login(self) -> None:
+        if not self.token:
+            self.login()
+
+    def request(
+        self,
+        method: str,
+        path: str,
+        *,
+        expected_status: int | tuple[int, ...] = 200,
+        **kwargs,
+    ) -> requests.Response:
+        self._ensure_login()
+
+        if isinstance(expected_status, int):
+            expected = (expected_status,)
+        else:
+            expected = expected_status
+
+        response = self.session.request(
+            method,
+            f"{self.api_url}{path}",
+            timeout=kwargs.pop("timeout", 15),
+            verify=self.config.proxy_verify_tls,
+            **kwargs,
+        )
+
+        if response.status_code not in expected:
+            raise RuntimeError(
+                f"NPM {method.upper()} {path} failed: "
+                f"HTTP {response.status_code}: "
+                f"{response.text[:300]}"
+            )
+
+        return response
+
+    def get_json(
+        self,
+        path: str,
+        *,
+        expected_status: int | tuple[int, ...] = 200,
+    ) -> Any:
+        response = self.request(
+            "GET",
+            path,
+            expected_status=expected_status,
+        )
+        return response.json()
+
+    def proxy_hosts(self) -> list[dict[str, Any]]:
+        data = self.get_json("/nginx/proxy-hosts")
+
+        if not isinstance(data, list):
+            raise RuntimeError(
+                "NPM proxy-host response was not a list"
+            )
+
+        return data