260c02f81d
- Enhanced the `vlans`, `interfaces`, and `system` properties with type hints for better clarity and type checking. - Added detailed docstrings to each property, describing the return types and structure of the associated configurations.
128 lines
4.0 KiB
Python
128 lines
4.0 KiB
Python
import json
|
|
from collections.abc import Iterator
|
|
from functools import cached_property
|
|
from typing import TYPE_CHECKING, Generic, TypeVar
|
|
|
|
from pydantic import BaseModel
|
|
|
|
from .interfaces import BaseDevice, device_registry
|
|
|
|
if TYPE_CHECKING:
|
|
from requests import Session
|
|
|
|
from oxi.interfaces.contract import Interfaces, System, Vlans
|
|
|
|
|
|
TModel = TypeVar("TModel", bound=BaseModel)
|
|
|
|
|
|
class ModelView(Generic[TModel]):
|
|
def __init__(self, model: TModel | list[TModel]):
|
|
self._model = model
|
|
|
|
def dump_json(self) -> str:
|
|
if isinstance(self._model, list):
|
|
return json.dumps(
|
|
[item.model_dump(by_alias=True) for item in self._model],
|
|
ensure_ascii=False,
|
|
)
|
|
return self._model.model_dump_json(by_alias=True)
|
|
|
|
def dump(self) -> dict | list:
|
|
if isinstance(self._model, list):
|
|
return [item.model_dump(by_alias=True) for item in self._model]
|
|
return self._model.model_dump(by_alias=True)
|
|
|
|
def __iter__(self) -> Iterator[TModel]:
|
|
if isinstance(self._model, list):
|
|
return iter(self._model)
|
|
raise TypeError("This view wraps a single model, not a list")
|
|
|
|
def __len__(self) -> int:
|
|
if isinstance(self._model, list):
|
|
return len(self._model)
|
|
raise TypeError("This view wraps a single model, not a list")
|
|
|
|
def __getitem__(self, item):
|
|
if isinstance(self._model, list):
|
|
return self._model[item]
|
|
raise TypeError("This view wraps a single model, not a list")
|
|
|
|
def __getattr__(self, item):
|
|
return getattr(self._model, item)
|
|
|
|
def __repr__(self) -> str:
|
|
return repr(self._model)
|
|
|
|
|
|
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.lower()
|
|
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, name=self._full_name).parse()
|
|
|
|
@cached_property
|
|
def _response(self):
|
|
response = self._session.get(self._url)
|
|
response.raise_for_status()
|
|
return response
|
|
|
|
@property
|
|
def text(self):
|
|
return self._response.text
|
|
|
|
def dump_json(self):
|
|
return self._parsed_data.model_dump_json(by_alias=True)
|
|
|
|
def dump(self):
|
|
return self._parsed_data.model_dump(by_alias=True)
|
|
|
|
def __str__(self):
|
|
return self.text
|
|
|
|
@property
|
|
def vlans(self) -> list["Vlans"]:
|
|
"""
|
|
Returns the list of VLAN configurations for the node.
|
|
|
|
:return: List of Vlans associated with the device.
|
|
vlan_id: int
|
|
name: str | None = Field(default=None, alias="description")
|
|
:rtype: list[oxi.interfaces.contract.Vlans]
|
|
"""
|
|
return ModelView(self._parsed_data.vlans)
|
|
return ModelView(self._parsed_data.vlans)
|
|
|
|
@property
|
|
def interfaces(self) -> list["Interfaces"]:
|
|
"""
|
|
Returns the list of Interfaces configurations for the node.
|
|
|
|
:return: List of Interfaces associated with the device.
|
|
name: str = Field(alias="interface")
|
|
ip_address: IPv4Address | None = None
|
|
mask: int | None = None
|
|
description: str | None = None
|
|
shutdown: bool = False
|
|
:rtype: list[oxi.interfaces.contract.Interfaces]
|
|
"""
|
|
return ModelView(self._parsed_data.interfaces)
|
|
|
|
@property
|
|
def system(self) -> "System":
|
|
"""
|
|
Returns the Systems configurations for the node.
|
|
|
|
:return: System associated with the device.
|
|
model: str
|
|
serial_number: str
|
|
version: str
|
|
:rtype: oxi.interfaces.contract.System
|
|
"""
|
|
return ModelView(self._parsed_data.system)
|