Add SuppressApiNotFound

helpy/helpy/__init__.py

@@ -20,6 +20,7 @@ from helpy._interfaces.time import OffsetTimeControl, SpeedUpRateTimeControl, St
 from helpy._interfaces.transaction_helper import Transaction
 from helpy._interfaces.url import HttpUrl, P2PUrl, WsUrl
 from helpy._sanitize import sanitize
+from helpy._suppress_api_not_found import SuppressApiNotFound
 
 __version__ = "0.0.0"
 
@@ -47,6 +48,7 @@ __all__ = [
     "Stopwatch",
     "SelfContextAsync",
     "SelfContextSync",
+    "SuppressApiNotFound",
     "sanitize",
     "Time",
     "TimeFormats",

helpy/helpy/_suppress_api_not_found.py

+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from helpy._interfaces.context import SelfContextSync
+from helpy.exceptions import ApiNotFoundError, GroupedErrorsError
+
+if TYPE_CHECKING:
+    from types import TracebackType
+
+
+class SuppressApiNotFound(SelfContextSync):
+    """Use to suppress ApiNotFoundError.
+
+    Example:
+    ```python
+
+    with SuppressApiNotFound("database_api", "rc_api") as suppressed:
+        node.api.database.get_config()
+        with node.batch() as bnode:
+            bnode.api.rc.get_resource_params()
+            bnode.api.jsonrpc.get_methods()
+
+    print("suppressed apis:", [x.api for x in suppressed.errors])
+    # suppressed apis: ['database_api', 'rc_api']
+    ```
+    """
+
+    def __init__(self, *apis: str) -> None:
+        super().__init__()
+        self.__apis = list(apis)
+        self.__suppressed_errors: list[ApiNotFoundError] = []
+
+    def _finally(self) -> None:
+        return
+
+    @property
+    def errors(self) -> list[ApiNotFoundError]:
+        return self.__suppressed_errors
+
+    def _handle_exception(self, exception: BaseException, __: TracebackType | None) -> bool:
+        if not isinstance(exception, ApiNotFoundError):
+            return False
+
+        cause = exception.cause
+        if (cause is None) or (not isinstance(cause, GroupedErrorsError)):
+            raise ValueError("Cannot access cause of exception to retrieve all information")
+
+        for ex in cause.exceptions:
+            if not isinstance(ex, ApiNotFoundError):
+                return False
+
+            if ex.api not in self.__apis:
+                return False
+
+            self.__suppressed_errors.append(ex)
+
+        return True

tests/helpy_test/functional/test_batch_node.py

 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Literal
+from typing import TYPE_CHECKING, Final, Literal
 
 import pytest
 
+from helpy import SuppressApiNotFound
 from helpy.exceptions import CommunicationError, NothingToSendError, ResponseNotReadyError
 
 if TYPE_CHECKING:
@@ -78,3 +79,17 @@ async def test_batch_node_nothing_to_send(sync_node: Hived) -> None:
     with pytest.raises(NothingToSendError):  # noqa: SIM117
         with sync_node.batch():
             pass
+
+
+def test_batch_node_with_suppress_api_not_found(sync_node: Hived) -> None:
+    # ARRANGE
+    missing_api: Final[str] = "debug_node_api"
+    amount_of_requests: Final[int] = 3
+
+    with SuppressApiNotFound(missing_api) as suppress, sync_node.batch() as bnode:
+        for _ in range(amount_of_requests):
+            bnode.api.debug_node.debug_get_head_block()
+
+    # ASSERT
+    assert len(suppress.errors) == amount_of_requests, "there should be exactly 2 suppressed errors"
+    assert all(item.api == missing_api for item in suppress.errors), f"suppressed for: {suppress.errors}"

tests/helpy_test/unit/test_suppress_api_not_found.py

+from __future__ import annotations
+
+import json
+from typing import cast
+
+import pytest
+
+from helpy import HttpUrl, SuppressApiNotFound
+from helpy._communication.rules import ApiNotFound
+from helpy.exceptions import ApiNotFoundError, GroupedErrorsError
+
+
+def api_not_found_error(api: str) -> ApiNotFoundError:
+    response = {
+        "jsonrpc": "2.0",
+        "error": {
+            "code": -32003,
+            "message": ("Assert Exception:api_itr != " "data._registered_apis.end(): Could not find API " + api),
+        },
+        "id": 1,
+    }
+
+    result = ApiNotFound(
+        url=HttpUrl("0.0.0.0:0"),
+        request={"jsonrpc": "2.0", "id": 1, "method": f"{api}.some_method"},
+    ).check(response=response, response_raw=json.dumps(response))
+    assert len(result) == 1, "Exception has not been generated"
+    return cast(ApiNotFoundError, result[0])
+
+
+@pytest.mark.parametrize(
+    "error",
+    [
+        api_not_found_error(api=api)
+        for api in [
+            "rc_api",
+            "database_api",
+            "account_history_api",
+            "future_plugin_that_not_exists_yet_api",
+        ]
+    ],
+)
+def test_suppress_api_not_found(error: ApiNotFoundError) -> None:
+    # ARRANGE & ACT
+    with SuppressApiNotFound(error.api) as suppressed:
+        raise error from GroupedErrorsError([error])
+
+    # ASSERT
+    assert suppressed.errors[0].api == error.api
+
+
+@pytest.mark.parametrize(
+    "error",
+    [
+        api_not_found_error("debug_node_api"),
+        ValueError("some value error"),
+    ],
+)
+def test_suppress_api_not_found_rethrow(error: Exception) -> None:
+    # ARRANGE
+
+    # ACT & ASSERT
+    with pytest.raises(type(error)), SuppressApiNotFound("rc_api", "database_api") as suppressed:
+        raise error from GroupedErrorsError([error])
+
+    assert len(suppressed.errors) == 0, "No errors should be suppressed"