Add ttp dependency and refactor OxiAPI and NodeConfig classes

- Added `ttp` as a dependency in `pyproject.toml` and `uv.lock`.
- Updated `NodeConfig` to store model names in lowercase.
- Refactored `OxiAPI` to always create a new session and added a `close` method.
- Removed unnecessary logging in `Node` class.
- Introduced interfaces for device registration with a new `BaseDevice` class and a `register_parser` function.
- Created initial structure for device models, including a `Mikrotik` parser.
This commit is contained in:
IluaAir
2026-02-14 21:31:02 +03:00
parent 8e85086d98
commit b60182ef3c
9 changed files with 57 additions and 19 deletions

View File

@@ -1,6 +1,7 @@
from functools import cached_property
from typing import TYPE_CHECKING
from .interfaces import BaseDevice, device_registry
if TYPE_CHECKING:
from requests import Session
@@ -10,7 +11,7 @@ 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._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:
@@ -19,7 +20,6 @@ class NodeConfig:
@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

View File

@@ -7,13 +7,12 @@ 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 = Session()
self._session.verify = verify
if username and password:
self._session.auth = (username, password)
@@ -23,13 +22,7 @@ class OxiAPI:
return self
def __exit__(self, *args):
self._session.close()
self.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()
def close(self):
return self._session.close()

View File

@@ -0,0 +1,21 @@
from typing import Callable, Type
from .base import BaseDevice
device_registry = {}
def register_parser(
name: list[str],
) -> Callable[[Type[BaseDevice]], Type[BaseDevice]]:
def wrapper(cls):
for item in name:
device_registry[item.lower()] = cls
return cls
return wrapper
from . import models # noqa: E402, F401
__all__ = ["register_parser", "device_registry"]

1
oxi/interfaces/base.py Normal file
View File

@@ -0,0 +1 @@
class BaseDevice: ...

View File

@@ -0,0 +1,7 @@
import importlib
import pkgutil
package = __package__
for loader, module_name, is_pkg in pkgutil.iter_modules(__path__):
importlib.import_module(f"{package}.{module_name}")

View File

@@ -0,0 +1,5 @@
from oxi.interfaces import register_parser
@register_parser(["routeros", "ros", "mikrotik"])
class Mikrotik: ...

View File

@@ -1,6 +1,6 @@
from typing import TYPE_CHECKING
from oxi.view import NodeView
from .view import NodeView
if TYPE_CHECKING:
@@ -19,9 +19,6 @@ class Node:
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()