131 lines
3.7 KiB
Python
131 lines
3.7 KiB
Python
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)
|
|
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()
|
|
)
|
|
|
|
|
|
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
|