diff --git a/src/delta_barth/errors.py b/src/delta_barth/errors.py index 2709698..83cb2ed 100644 --- a/src/delta_barth/errors.py +++ b/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 diff --git a/src/delta_barth/types.py b/src/delta_barth/types.py index 5b25a70..9399ff2 100644 --- a/src/delta_barth/types.py +++ b/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], }, ) diff --git a/tests/test_errors.py b/tests/test_errors.py index 74a1d75..9844c84 100644 --- a/tests/test_errors.py +++ b/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