"""Public OPNsense client class composed from domain mixins."""
from collections.abc import Awaitable, Callable
from types import TracebackType
from typing import Self, TypeVar
import aiohttp
from .client_base import ClientBaseMixin
from .const import OPNSENSE_LTD_FIRMWARE, OPNSENSE_MIN_FIRMWARE
from .dhcp import DHCPMixin
from .exceptions import (
OPNsenseBelowMinFirmware,
OPNsenseUnknownFirmware,
_map_opnsense_exception,
)
from .firewall import FirewallMixin
from .firmware import FirmwareMixin
from .helpers import _LOGGER, firmware_is_at_least
from .nut import NutMixin
from .services import ServicesMixin
from .smart import SmartMixin
from .speedtest import SpeedtestMixin
from .system import SystemMixin
from .telemetry import TelemetryMixin
from .traffic import TrafficMixin
from .unbound import UnboundMixin
from .vnstat import VnstatMixin
from .vouchers import VouchersMixin
from .vpn import VPNMixin
_T = TypeVar("_T")
[docs]
class OPNsenseClient(
ClientBaseMixin,
FirmwareMixin,
FirewallMixin,
DHCPMixin,
ServicesMixin,
SmartMixin,
NutMixin,
SpeedtestMixin,
SystemMixin,
UnboundMixin,
VouchersMixin,
TelemetryMixin,
TrafficMixin,
VnstatMixin,
VPNMixin,
):
"""Async client for supported OPNsense REST endpoints."""
async def __aenter__(self) -> Self:
"""Validate the client before entering an async context manager.
Returns:
Self: Validated client instance.
"""
await self.validate()
return self
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc: BaseException | None,
traceback: TracebackType | None,
) -> None:
"""Close background resources when leaving an async context manager.
Args:
exc_type (type[BaseException] | None): Exception type raised in the context block.
exc (BaseException | None): Exception instance raised in the context block.
traceback (TracebackType | None): Traceback for an exception raised in the context block.
"""
del exc_type, exc, traceback
await self.async_close()
async def _run_validation_request(self, request: Callable[[], Awaitable[_T]]) -> _T:
"""Run a validation request and map transport failures to public exceptions.
Args:
request (Callable[[], Awaitable[_T]]): Zero-argument async request
used during client validation.
Returns:
_T: Decoded response from the successful validation request.
Raises:
_map_opnsense_exception: Raised as the mapped public OPNsense error
when the request encounters an aiohttp client error or times out.
"""
try:
return await request()
except (aiohttp.ClientError, TimeoutError) as e:
raise _map_opnsense_exception(e) from e
[docs]
async def validate(self, *, require_device_id: bool = True) -> None:
"""Validate connectivity, authentication, and minimum firmware support.
This raises request failures regardless of ``self._throw_errors``; those
failures are mapped to public OPNsense errors by ``_run_validation_request``.
Args:
require_device_id (bool): Whether validation must resolve a physical-device
unique ID.
Raises:
OPNsenseUnknownFirmware: Raised when firmware detection returns no version.
OPNsenseBelowMinFirmware: Raised when the detected firmware is unsupported.
"""
orig_throw_errors = self._throw_errors
self._throw_errors = True
try:
if require_device_id:
fw_ver = await self._run_validation_request(self.get_host_firmware_version)
else:
await self._run_validation_request(self._store_host_firmware_version)
fw_ver = self._firmware_version
if fw_ver is None:
raise OPNsenseUnknownFirmware
meets_min_firmware = firmware_is_at_least(fw_ver, OPNSENSE_MIN_FIRMWARE)
if meets_min_firmware is None:
raise OPNsenseUnknownFirmware
if not meets_min_firmware:
msg = (
f"OPNsense Firmware {fw_ver} detected. "
f"aiopnsense requires OPNsense Firmware >= {OPNSENSE_MIN_FIRMWARE}"
)
raise OPNsenseBelowMinFirmware(msg)
meets_recommended_firmware = firmware_is_at_least(fw_ver, OPNSENSE_LTD_FIRMWARE)
if meets_recommended_firmware is None:
raise OPNsenseUnknownFirmware
if not meets_recommended_firmware:
_LOGGER.warning(
"OPNsense Firmware of %s is below the recommended >= %s. aiopnsense will work, but there may be some missing features.",
fw_ver,
OPNSENSE_LTD_FIRMWARE,
)
if require_device_id:
await self._run_validation_request(self.get_device_unique_id)
finally:
self._throw_errors = orig_throw_errors