Reconcile optional endpoint availability from real requests

aiopnsense/_typing.py

@@ -1,17 +1,32 @@
 """Typing protocol contracts for aiopnsense mixins."""
 
+import asyncio
 from collections.abc import AsyncGenerator, MutableMapping
 from datetime import tzinfo
-from typing import Any, Protocol
+from typing import Any, Literal, Protocol
 
 
 class AiopnsenseClientProtocol(Protocol):
     """Structural typing contract used by split aiopnsense mixins."""
 
     _throw_errors: bool
+    _endpoint_availability: dict[tuple[Literal["get", "post"], str], bool]
+    _endpoint_checked_at: dict[tuple[Literal["get", "post"], str], float]
+    _endpoint_locks: dict[tuple[Literal["get", "post"], str], asyncio.Lock]
+    _optional_endpoint_missing_pending_confirmation: set[tuple[Literal["get", "post"], str]]
 
     async def _get(self, path: str) -> MutableMapping[str, Any] | list | None: ...
 
+    async def _get_optional(
+        self, path: str
+    ) -> tuple[Literal["available", "malformed", "missing", "unavailable"], object]: ...
+
+    async def _post_optional(
+        self,
+        path: str,
+        payload: MutableMapping[str, Any] | None = None,
+    ) -> tuple[Literal["available", "malformed", "missing", "unavailable"], object]: ...
+
     async def _get_text(self, path: str) -> str | None: ...
 
     async def _post(
@@ -57,3 +72,19 @@ async def _is_get_endpoint_available(self, path: str, force_refresh: bool = Fals
     async def _is_post_endpoint_available(
         self, path: str, force_refresh: bool = False
     ) -> bool | None: ...
+
+    async def _check_optional_get_endpoint(
+        self,
+        path: str,
+        *,
+        force_refresh: bool = False,
+    ) -> tuple[Literal["available", "malformed", "missing", "unavailable"], object]: ...
+
+    async def _check_optional_post_endpoint(
+        self,
+        path: str,
+        payload: MutableMapping[str, Any] | None = None,
+        cache_path: str | None = None,
+        *,
+        force_refresh: bool = False,
+    ) -> tuple[Literal["available", "malformed", "missing", "unavailable"], object]: ...

aiopnsense/client_base.py

@@ -2,8 +2,7 @@
 
 import asyncio
 from collections.abc import MutableMapping
-from datetime import datetime
-from typing import Any
+from typing import Any, Literal
 from urllib.parse import urlparse
 import warnings
 
@@ -12,7 +11,7 @@
 from .client_endpoint import ClientEndpointMixin
 from .client_queue import ClientQueueMixin
 from .client_transport import ClientTransportMixin
-from .const import DEFAULT_CACHE_TTL_SECONDS
+from .const import DEFAULT_CACHE_TTL_SECONDS, DEFAULT_NEGATIVE_CACHE_TTL_SECONDS
 from .exceptions import OPNsenseInvalidArgument
 
 _UNSET: object = object()
@@ -41,7 +40,8 @@ def __init__(
             password (str): Password for API authentication.
             session (aiohttp.ClientSession): HTTP client session used for API requests.
             opts (MutableMapping[str, Any] | None, optional): Optional client configuration values
-                (e.g. ``opts={"verify_ssl": True}``).
+                such as ``verify_ssl``, ``endpoint_positive_cache_ttl_seconds``,
+                and ``endpoint_negative_cache_ttl_seconds``.
             initial (bool | object): Deprecated alias for ``throw_errors``. When provided,
                 a ``DeprecationWarning`` is emitted. Ignored when ``throw_errors`` is also set.
             throw_errors (bool | object): Whether request and decorator errors should be
@@ -79,9 +79,28 @@ def __init__(
                 self._throw_errors = initial
         self._firmware_version: str | None = None
         self._use_snake_case: bool | None = None
-        self._endpoint_availability: dict[str, bool] = {}
-        self._endpoint_checked_at: dict[str, datetime] = {}
-        self._endpoint_cache_ttl_seconds = DEFAULT_CACHE_TTL_SECONDS
+        self._endpoint_availability: dict[tuple[Literal["get", "post"], str], bool] = {}
+        self._endpoint_checked_at: dict[tuple[Literal["get", "post"], str], float] = {}
+        self._endpoint_locks: dict[tuple[Literal["get", "post"], str], asyncio.Lock] = {}
+        self._optional_endpoint_missing_pending_confirmation: set[
+            tuple[Literal["get", "post"], str]
+        ] = set()
+        positive_ttl = self._opts.get(
+            "endpoint_positive_cache_ttl_seconds", DEFAULT_CACHE_TTL_SECONDS
+        )
+        negative_ttl = self._opts.get(
+            "endpoint_negative_cache_ttl_seconds", DEFAULT_NEGATIVE_CACHE_TTL_SECONDS
+        )
+        if not isinstance(positive_ttl, int) or isinstance(positive_ttl, bool) or positive_ttl <= 0:
+            raise OPNsenseInvalidArgument(
+                "`endpoint_positive_cache_ttl_seconds` must be a positive integer."
+            )
+        if not isinstance(negative_ttl, int) or isinstance(negative_ttl, bool) or negative_ttl <= 0:
+            raise OPNsenseInvalidArgument(
+                "`endpoint_negative_cache_ttl_seconds` must be a positive integer."
+            )
+        self._endpoint_cache_ttl_seconds = positive_ttl
+        self._endpoint_negative_cache_ttl_seconds = negative_ttl
         self._rest_api_query_count = 0
         self._request_queue: asyncio.Queue = asyncio.Queue()
         self._workers: list[asyncio.Task[Any]] = []