import logging from functools import cached_property import requests from typing import Optional from oxi.interface.base import BaseDevice from settings import settings import oxi.interface.models # noqa from oxi.interface import device_registry log = logging.getLogger() class OxidizedAPI: def __init__( self, session: Optional[requests.Session] = None, username: Optional[str] = None, password: Optional[str] = None, verify: bool = True ): self.base_url = settings.oxi_url.rstrip('/') self._session = session or requests.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() class Node: def __init__(self, session: requests.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) return NodeView( session=self._session, base_url=self._base_url, data=response.json() ) class NodeView: def __init__(self, session: requests.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') @property def config(self): return NodeConfig(self._session, self.full_name, self.model, self._base_url) class NodeConfig: def __init__(self, session: requests.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._data = None @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._device(self.text).parse_config().vlans def l3interfaces(self): return self._device(self.text).parse_config().l3interfaces def vlaninterfaces(self): return self._device(self.text).parse_config().vlaninterfaces