add VscaleDomains output provider

pddnsc/outputs/vscale.py

@@ -0,0 +1,142 @@
+import os
+import httpx
+
+from typing import Optional
+
+from pddnsc.base import BaseOutputProvider
+
+
+class VscaleDomains(BaseOutputProvider):
+    def post_init(self):
+        self.__token = self.get_api_token()
+        if not self.__token:
+            raise KeyError("no api token, use env VSCALE_API_TOKEN")
+
+    def get_api_token(self) -> str:
+        token_env = self.config.get("api_token_env", "VSCALE_API_TOKEN")
+        return os.environ[token_env]
+
+    async def find_domain_id(self, transport) -> Optional[int]:
+        domain = self.config["domain"]
+        headers = {"X-Token": self.__token}
+        async with httpx.AsyncClient(transport=transport) as client:
+            response = await client.get(
+                "https://api.vscale.io/v1/domains/", headers=headers
+            )
+            if httpx.codes.is_success(response.status_code):
+                data = response.json()
+                if isinstance(data, list):
+                    for entry in data:
+                        if entry["name"] == domain:
+                            return entry["id"]
+                else:
+                    raise TypeError(
+                        "failed to find domain id, unexpected response type"
+                    )
+            else:
+                raise ValueError(
+                    f"failed to find domain id, code: {response.status_code}"
+                )
+
+    async def find_record(self, transport, domain_id, record_type) -> Optional[int]:
+        target = self.config["target"]
+        domain = self.config["domain"]
+        target = f"{target}.{domain}"
+        headers = {"X-Token": self.__token}
+        async with httpx.AsyncClient(transport=transport) as client:
+            response = await client.get(
+                f"https://api.vscale.io/v1/domains/{domain_id}/records/",
+                headers=headers,
+            )
+            if httpx.codes.is_success(response.status_code):
+                data = response.json()
+                if isinstance(data, list):
+                    for entry in data:
+                        if entry["name"] == target and entry["type"] == record_type:
+                            return entry["id"]
+            else:
+                raise RuntimeError(
+                    f"error list records {domain_id=}: ", response.status_code
+                )
+
+    async def get_record_value(self, transport, domain_id, record_id) -> str:
+        headers = {"X-Token": self.__token}
+        async with httpx.AsyncClient(transport=transport) as client:
+            response = await client.get(
+                f"https://api.vscale.io/v1/domains/{domain_id}/records/{record_id}",
+                headers=headers,
+            )
+            if httpx.codes.is_success(response.status_code):
+                data = response.json()
+                if isinstance(data, dict):
+                    return data["content"]
+
+    async def change_record(self, transport, domain_id, record_id, record_type, value):
+        target = self.config["target"]
+        domain = self.config["domain"]
+        target = f"{target}.{domain}"
+        ttl = self.config.get("ttl", 300)
+        headers = {"X-Token": self.__token, "Content-Type": "application/json"}
+        data = {
+            "content": value,
+            "name": target,
+            "ttl": ttl,
+            "type": record_type,
+            "id": record_id,
+        }
+        async with httpx.AsyncClient(transport=transport) as client:
+            response = await client.put(
+                f"https://api.vscale.io/v1/domains/{domain_id}/records/{record_id}",
+                headers=headers,
+                json=data,
+            )
+            if not httpx.codes.is_success(response.status_code):
+                raise RuntimeError(
+                    f"failed to change record: {target=},{domain_id=}, {record_id=}, {record_type=}, {value=}"
+                )
+
+    async def create_record(self, transport, domain_id, record_type, value):
+        target = self.config["target"]
+        domain = self.config["domain"]
+        target = f"{target}.{domain}"
+        ttl = self.config.get("ttl", 300)
+        headers = {"X-Token": self.__token, "Content-Type": "application/json"}
+        data = {"content": value, "name": target, "ttl": ttl, "type": record_type}
+        async with httpx.AsyncClient(transport=transport) as client:
+            response = await client.post(
+                f"https://api.vscale.io/v1/domains/{domain_id}/records/",
+                headers=headers,
+                json=data,
+            )
+            if not httpx.codes.is_success(response.status_code):
+                raise RuntimeError(
+                    f"failed to create record: {target=},{domain_id=}, {record_type=}, {value=}, {response.status_code=}"
+                )
+
+    async def set_addrs_imp(self, source_provider, addr_v4, addr_v6):
+        transport = self.best_transport(addr_v4, addr_v6)
+        domain_id = await self.find_domain_id(transport)
+        save_ipv4 = self.config.get("ipv4", False)
+        save_ipv6 = self.config.get("ipv6", False)
+        if "ipv4" not in self.config and "ipv6" not in self.config:
+            save_ipv4 = save_ipv6 = True
+
+        save_addrs = []
+        if addr_v4 and save_ipv4:
+            save_addrs.append(("A", addr_v4))
+
+        if addr_v6 and save_ipv6:
+            save_addrs.append(("AAAA", addr_v6))
+
+        for record_type, value in save_addrs:
+            record_id = await self.find_record(transport, domain_id, record_type)
+            if record_id:
+                old_value = await self.get_record_value(transport, domain_id, record_id)
+                if old_value != value:
+                    await self.change_record(
+                        transport, domain_id, record_id, record_type, value
+                    )
+                else:
+                    print(f"vscale: skip record change ({record_type=}), value equal")
+            else:
+                await self.create_record(transport, domain_id, record_type, value)