base structure

oxi/__init__.py

@@ -0,0 +1,6 @@
+from .core import OxiAPI
+
+
+__all__ = [
+    "OxiAPI",
+]

oxi/conf.py

@@ -0,0 +1,45 @@
+from functools import cached_property
+from typing import TYPE_CHECKING
+
+
+if TYPE_CHECKING:
+    from requests import Session
+
+
+class NodeConfig:
+    def __init__(self, session: "Session", full_name: str, model: str, base_url: str):
+        self._session = session
+        self._full_name = full_name
+        self._model = model
+        self._url = f"{base_url}/node/fetch/{full_name}"
+        self._device: type[BaseDevice] = device_registry.get(self._model.lower())
+        if self._device is None:
+            raise ValueError(f"Device model '{self._model}' not found in registry")
+        self._parsed_data = self._device(self.text).parse_config()
+
+    @cached_property
+    def _response(self):
+        log.debug(f"Fetching config from {self._url}")
+        response = self._session.get(self._url)
+        response.raise_for_status()
+        return response
+
+    @property
+    def text(self):
+        return self._response.text
+
+    @property
+    def json(self):
+        return self._response.json()
+
+    def __str__(self):
+        return self.text
+
+    def vlans(self):
+        return self._parsed_data.vlans
+
+    def l3interfaces(self):
+        return self._parsed_data.l3interfaces
+
+    def vlaninterfaces(self):
+        return self._parsed_data.vlaninterfaces

oxi/core.py

@@ -0,0 +1,35 @@
+from typing import Optional
+from requests import Session
+from .node import Node
+
+
+class OxiAPI:
+    def __init__(
+        self,
+        url: str,
+        session: Optional[Session] = None,
+        username: Optional[str] = None,
+        password: Optional[str] = None,
+        verify: bool = True,
+    ):
+        self.base_url = url.rstrip("/")
+        self._session = session or Session()
+        self._session.verify = verify
+        if username and password:
+            self._session.auth = (username, password)
+        self.node = Node(self._session, self.base_url)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        self._session.close()
+
+    def get(self, endpoint: str, **kwargs) -> dict:
+        url = f"{self.base_url}/{endpoint.lstrip('/')}"
+        if not url.endswith(".json"):
+            url += ".json"
+        result = self._session.get(url, **kwargs)
+        if result.status_code == 500:
+            raise ValueError(f"page {url} not found")
+        return result.json()

oxi/node.py

@@ -0,0 +1,28 @@
+from typing import TYPE_CHECKING
+
+from oxi.view import NodeView
+
+
+if TYPE_CHECKING:
+    from requests import Session
+
+
+class Node:
+    def __init__(self, session: "Session", base_url: str):
+        self._session = session
+        self._base_url = base_url
+        self._data = None
+
+    def __call__(self, name: str) -> NodeView:
+        url = f"{self._base_url}/node/show/{name}"
+        if not url.endswith(".json"):
+            url += ".json"
+        response = self._session.get(url)
+        if response.status_code == 500:
+            log.warning(
+                "Oxidized response: %r , %r not found", response.status_code, url
+            )
+            raise ValueError(f"page {url} not found")
+        return NodeView(
+            session=self._session, base_url=self._base_url, data=response.json()
+        )

oxi/view.py

@@ -0,0 +1,35 @@
+from functools import cached_property
+from typing import TYPE_CHECKING
+
+from .conf import NodeConfig
+
+
+if TYPE_CHECKING:
+    from requests import Session
+
+
+class NodeView:
+    def __init__(self, session: "Session", base_url: str, data: dict):
+        self._session = session
+        self._base_url = base_url
+        self._data = data
+
+    @property
+    def ip(self):
+        return self._data.get("ip")
+
+    @property
+    def full_name(self):
+        return self._data.get("full_name")
+
+    @property
+    def group(self):
+        return self._data.get("group")
+
+    @property
+    def model(self):
+        return self._data.get("model")
+
+    @cached_property
+    def config(self):
+        return NodeConfig(self._session, self.full_name, self.model, self._base_url)