Initial AS213905 SDK release

python/src/as213905/__init__.py

@@ -0,0 +1,4 @@
+from .client import AS213905, APIError
+
+__all__ = ["AS213905", "APIError"]
+__version__ = "1.1.0"

python/src/as213905/client.py

@@ -0,0 +1,99 @@
+"""Typed, dependency-free AS213905 API client."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from typing import Any, Iterable, Mapping, Optional
+from urllib.error import HTTPError, URLError
+from urllib.parse import quote, urlencode
+from urllib.request import Request, urlopen
+
+
+@dataclass
+class APIError(Exception):
+    status: int
+    code: str
+    message: str = ""
+    required_scope: Optional[str] = None
+    retry_after: Optional[int] = None
+
+    def __str__(self) -> str:
+        detail = f": {self.message}" if self.message else ""
+        return f"AS213905 API error {self.status} ({self.code}){detail}"
+
+
+class AS213905:
+    def __init__(self, api_key: str, *, base_url: str = "https://as213905.com", timeout: float = 30.0) -> None:
+        if not api_key.strip():
+            raise ValueError("api_key is required")
+        self.api_key = api_key
+        self.base_url = base_url.rstrip("/")
+        self.timeout = timeout
+
+    def _request(self, method: str, path: str, *, organization_id: Optional[str] = None, body: Any = None) -> Any:
+        payload = None if body is None else json.dumps(body).encode()
+        headers = {"Authorization": f"Bearer {self.api_key}", "Accept": "application/json", "User-Agent": "as213905-python/1.1.0"}
+        if payload is not None:
+            headers["Content-Type"] = "application/json"
+        if organization_id:
+            headers["X-Organization-ID"] = organization_id
+        request = Request(self.base_url + path, data=payload, headers=headers, method=method)
+        try:
+            with urlopen(request, timeout=self.timeout) as response:
+                raw = response.read()
+        except HTTPError as exc:
+            try:
+                value = json.loads(exc.read())
+            except (ValueError, UnicodeDecodeError):
+                value = {}
+            retry = value.get("retry_after") or exc.headers.get("Retry-After")
+            raise APIError(exc.code, value.get("error", "http_error"), value.get("message", ""), value.get("required_scope"), int(retry) if retry else None) from None
+        except URLError as exc:
+            raise APIError(0, "network_error", str(exc.reason)) from exc
+        return json.loads(raw) if raw else None
+
+    @staticmethod
+    def _traffic_query(days: Optional[int], from_date: Optional[str], to_date: Optional[str]) -> str:
+        query = {key: value for key, value in (("days", days), ("from", from_date), ("to", to_date)) if value is not None}
+        return "?" + urlencode(query) if query else ""
+
+    def organization(self) -> Mapping[str, Any]:
+        return self._request("GET", "/api/v1/organization")["organization"]
+    def invoices(self) -> list[Mapping[str, Any]]:
+        return self._request("GET", "/api/v1/invoices")["invoices"]
+    def services(self) -> list[Mapping[str, Any]]:
+        return self._request("GET", "/api/v1/services")["services"]
+    def service(self, service_id: str) -> Mapping[str, Any]:
+        return self._request("GET", f"/api/v1/services/{quote(service_id, safe='')}" )["service"]
+    def service_metrics(self, service_id: str, *, target: Optional[str] = None, range: Optional[str] = None) -> Mapping[str, Any]:
+        query = urlencode({k: v for k, v in (("target", target), ("range", range)) if v is not None})
+        return self._request("GET", f"/api/v1/services/{quote(service_id, safe='')}/metrics" + ("?" + query if query else ""))
+    def power_service(self, service_id: str, action: str) -> Mapping[str, Any]:
+        return self._request("POST", f"/api/v1/services/{quote(service_id, safe='')}/power", body={"action": action})
+    def traffic(self, *, days: Optional[int] = None, from_date: Optional[str] = None, to_date: Optional[str] = None) -> Mapping[str, Any]:
+        return self._request("GET", "/api/v1/traffic" + self._traffic_query(days, from_date, to_date))
+    def tunnels(self) -> list[Mapping[str, Any]]:
+        return self._request("GET", "/api/v1/tunnels")["tunnels"]
+    def create_tunnel(self, peer_endpoint: str, location: str, *, label: Optional[str] = None, announce: bool = False) -> Mapping[str, Any]:
+        body = {"peer_endpoint": peer_endpoint, "location": location, "announce": announce}
+        if label is not None: body["label"] = label
+        return self._request("POST", "/api/v1/tunnels", body=body)["tunnel"]
+    def delete_tunnel(self, tunnel_id: str) -> None:
+        self._request("DELETE", f"/api/v1/tunnels/{quote(tunnel_id, safe='')}")
+    def tunnel_traffic(self, tunnel_id: str, *, days: Optional[int] = None, from_date: Optional[str] = None, to_date: Optional[str] = None) -> Mapping[str, Any]:
+        return self._request("GET", f"/api/v1/tunnels/{quote(tunnel_id, safe='')}/traffic" + self._traffic_query(days, from_date, to_date))
+    def geofeeds(self, organization_id: str) -> Mapping[str, Any]:
+        return self._request("GET", "/api/v1/geofeeds", organization_id=organization_id)
+    def create_geofeed(self, organization_id: str, name: str) -> Mapping[str, Any]:
+        return self._request("POST", "/api/v1/geofeeds", organization_id=organization_id, body={"name": name})
+    def geofeed(self, organization_id: str, geofeed_id: str) -> Mapping[str, Any]:
+        return self._request("GET", f"/api/v1/geofeeds/{quote(geofeed_id, safe='')}", organization_id=organization_id)["geofeed"]
+    def rename_geofeed(self, organization_id: str, geofeed_id: str, name: str) -> None:
+        self._request("PATCH", f"/api/v1/geofeeds/{quote(geofeed_id, safe='')}", organization_id=organization_id, body={"name": name})
+    def delete_geofeed(self, organization_id: str, geofeed_id: str) -> None:
+        self._request("DELETE", f"/api/v1/geofeeds/{quote(geofeed_id, safe='')}", organization_id=organization_id)
+    def replace_geofeed_records(self, organization_id: str, geofeed_id: str, records: Iterable[Mapping[str, Any]]) -> None:
+        self._request("PUT", f"/api/v1/geofeeds/{quote(geofeed_id, safe='')}/records", organization_id=organization_id, body={"records": list(records)})
+    def rotate_geofeed_url(self, organization_id: str, geofeed_id: str) -> Mapping[str, Any]:
+        return self._request("POST", f"/api/v1/geofeeds/{quote(geofeed_id, safe='')}/rotate", organization_id=organization_id)