prepare results wrapping for error handling architecture

src/delta_barth/errors.py

@@ -1,13 +1,19 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any, Final
+import typing as t
+from collections.abc import Callable
+from functools import wraps
+from typing import Any, Final
 
 from delta_barth.constants import DEFAULT_API_ERR_CODE, DEFAULT_INTERNAL_ERR_CODE
 from delta_barth.types import DataPipeStates, Status
 
-if TYPE_CHECKING:
+if t.TYPE_CHECKING:
     from delta_barth.types import DelBarApiError, StatusDescription
 
+P = t.ParamSpec("P")
+T = t.TypeVar("T")
+
 
 class UnspecifiedRequestType(Exception):
     """exception raised if for a given API endpoint a not defined operation is requested"""
@@ -21,6 +27,12 @@ class FeaturesMissingError(Exception):
     """exception raised if needed features are missing"""
 
 
+# ** Exceptions for result wrappers
+class WAccessResultDespiteError(Exception):
+    """wrapped results exception: raised if result is accessed, even though
+    there was an error in the underlying procedure"""
+
+
 # ** Exceptions for unwrap of error value
 class UDataProcessingError(Exception):
     """unwrap exception: all data related errors (e.g. wrong format, non-sufficient quality)"""
@@ -88,8 +100,17 @@ class StatusHandler:
 
         self._pipe_states = DataPipeStates(**parsed_errors)
 
-    def error(
+    def exception_to_error(
         self,
+        exception: Exception,
+        code: int = DEFAULT_INTERNAL_ERR_CODE,
+    ) -> Status:
+        description = exception.__class__.__name__
+        message = str(exception)
+        return self.error(description, message, code)
+
+    @staticmethod
+    def error(
         description: str,
         message: str = "",
         code: int = DEFAULT_INTERNAL_ERR_CODE,
@@ -108,8 +129,8 @@ class StatusHandler:
             message=message,
         )
 
+    @staticmethod
     def api_error(
-        self,
         error: DelBarApiError,
     ) -> Status:
         description = "Es ist ein Fehler bei der Kommunikation mit dem API-Server aufgetreten"
@@ -148,3 +169,68 @@ class StatusHandler:
 
 
 STATUS_HANDLER: Final[StatusHandler] = StatusHandler()
+
+
+# ** result wrapping
+class NotSet:
+    __slots__ = tuple()
+
+    def __init__(self) -> None: ...
+
+    def __str__(self) -> str:
+        return ">Not set<"
+
+    def __repr__(self) -> str:
+        return f"{self.__class__.__name__}()"
+
+
+class ResultWrapper(t.Generic[T]):
+    def __init__(
+        self,
+        result: T | NotSet,
+        exception: Exception | None,
+        code_on_error: int,
+    ) -> None:
+        self._result = result
+        status: Status = STATUS_HANDLER.SUCCESS
+        if exception is not None:
+            status = STATUS_HANDLER.exception_to_error(exception, code=code_on_error)
+        self.status = status
+
+    @property
+    def result(self) -> T:
+        if isinstance(self._result, NotSet):
+            raise WAccessResultDespiteError(
+                "Can not access result because an error occurred during procedure"
+            )
+        return self._result
+
+    def __str__(self) -> str:
+        return f"Status(result: {self._result}, status: {self.status})"
+
+    def __repr__(self) -> str:
+        return self.__str__()
+
+
+def wrap_result(
+    code_on_error: int = DEFAULT_API_ERR_CODE,
+) -> Callable[[Callable[P, T]], Callable[P, ResultWrapper[T]]]:
+    def wrap_result(func: Callable[P, T]) -> Callable[P, ResultWrapper[T]]:
+        @wraps(func)
+        def wrapper(*args: P.args, **kwargs: P.kwargs) -> ResultWrapper[T]:
+            status: ResultWrapper[T]
+            try:
+                res = func(*args, **kwargs)
+                status = ResultWrapper(
+                    result=res, exception=None, code_on_error=code_on_error
+                )
+            except Exception as err:
+                status = ResultWrapper(
+                    result=NotSet(), exception=err, code_on_error=code_on_error
+                )
+
+            return status
+
+        return wrapper
+
+    return wrap_result

src/delta_barth/types.py

@@ -1,14 +1,20 @@
 from __future__ import annotations
 
 import enum
+import typing as t
 from dataclasses import dataclass, field
-from typing import NotRequired, TypeAlias, TypedDict
 
 import pandas as pd
 from pydantic import BaseModel, ConfigDict, SkipValidation
 
 # ** Pipeline state management
-StatusDescription: TypeAlias = tuple[str, int, str]
+StatusDescription: t.TypeAlias = tuple[str, int, str]
+
+
+class IError(t.Protocol):
+    code: int
+    description: str
+    message: str
 
 
 class Status(BaseModel):
@@ -59,12 +65,12 @@ class HttpRequestTypes(enum.StrEnum):
     DELETE = enum.auto()
 
 
-HttpContentHeaders = TypedDict(
+HttpContentHeaders = t.TypedDict(
     "HttpContentHeaders",
     {
         "Content-type": str,
         "Accept": str,
-        "DelecoToken": NotRequired[str],
+        "DelecoToken": t.NotRequired[str],
     },
 )
 

tests/test_errors.py

@@ -76,7 +76,7 @@ def test_status_handler_api_error():
     assert new_err.api_server_error == api_err
 
 
-def test_status_handler_raising():
+def test_status_handler_unwrap():
     status_hdlr = errors.StatusHandler()
 
     # success: should not raise
@@ -121,3 +121,47 @@ def test_status_handler_raising():
             assert description in descr
             assert msg in descr
             raise err
+
+
+def test_status_handler_exception_parsing():
+    status_hdlr = errors.StatusHandler()
+    test_message = "Test Exception"
+    test_code = 110
+    exception = ValueError(test_message)
+    status = status_hdlr.exception_to_error(exception, test_code)
+    assert status.code == test_code
+    assert status.description == exception.__class__.__name__
+    assert status.message == test_message
+
+
+def test_not_set():
+    not_set = errors.NotSet()
+    # using slots, dynamic attribute generation should not be possible
+    with pytest.raises(AttributeError):
+        not_set.test = "try to set value"  # type: ignore
+
+
+def test_result_wrapper_class():
+    test_result = 10
+    error_code = 146
+    twrapper: errors.ResultWrapper[int] = errors.ResultWrapper(
+        result=test_result,
+        exception=None,
+        code_on_error=error_code,
+    )
+    assert twrapper.status == errors.STATUS_HANDLER.SUCCESS
+    assert twrapper.result == test_result
+    assert twrapper.status.code != error_code
+    # test for no result
+    test_result = errors.NotSet()
+    test_message = "Test of error message"
+    exception = ValueError(test_message)
+    twrapper: errors.ResultWrapper[int] = errors.ResultWrapper(
+        result=test_result,
+        exception=exception,
+        code_on_error=error_code,
+    )
+    assert twrapper.status != errors.STATUS_HANDLER.SUCCESS
+    assert twrapper.status.code == error_code
+    with pytest.raises(errors.WAccessResultDespiteError):
+        twrapper.result