diff --git a/pyproject.toml b/pyproject.toml index 4f4bfca..a5427e1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "dopt-basics" -version = "0.1.3" +version = "0.1.4dev1" description = "basic cross-project tools for Python-based d-opt projects" authors = [ {name = "Florian Förster", email = "f.foerster@d-opt.com"}, @@ -69,7 +69,7 @@ directory = "reports/coverage" [tool.bumpversion] -current_version = "0.1.3" +current_version = "0.1.4dev1" parse = """(?x) (?P0|[1-9]\\d*)\\. (?P0|[1-9]\\d*)\\. diff --git a/scripts/bump_patch.ps1 b/scripts/bump_patch.ps1 new file mode 100644 index 0000000..19ee4d1 --- /dev/null +++ b/scripts/bump_patch.ps1 @@ -0,0 +1,2 @@ +pdm run bump-my-version bump patch +pdm run bump-my-version show current_version \ No newline at end of file diff --git a/src/dopt_basics/result_pattern.py b/src/dopt_basics/result_pattern.py new file mode 100644 index 0000000..9a2830a --- /dev/null +++ b/src/dopt_basics/result_pattern.py @@ -0,0 +1,201 @@ +from __future__ import annotations + +import dataclasses as dc +import inspect +import typing as t +from collections.abc import Callable +from functools import wraps +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from logging import Logger + +P = t.ParamSpec("P") +T = t.TypeVar("T") + + +# ** Exceptions for result wrappers +class WrapperAccessResultDespiteError(Exception): + """wrapped results exception: raised if result is accessed, even though + there was an error in the underlying procedure""" + + +@dc.dataclass(kw_only=True, slots=True) +class Status: + code: int + description: str + message: str + ExceptionType: type[Exception] | None = None + + +class StatusHandler: + __slots__ = ("logger", "_SUCCESS") + + def __init__( + self, + logger: Logger | None = None, + ) -> None: + self.logger = logger + + self._SUCCESS: t.Final[Status] = Status( + code=0, description="SUCCESS", message="operation executed successfully" + ) + + @property + def SUCCESS(self) -> Status: + return self._SUCCESS + + def error_to_exception( + self, + error_state: Status, + ) -> Exception: + if error_state.ExceptionType is None: + raise ValueError( + "Cannot construct exception from state where no exception is defined" + ) + return error_state.ExceptionType(error_state.message) + + def exception_to_error( + self, + exception: Exception, + code: int, + ) -> Status: + doc_string = inspect.getdoc(exception) + description = doc_string if doc_string is not None else "" + message = str(exception) + return self.error_state( + code, + description, + message, + exception=exception, + ) + + @staticmethod + def error_state( + code: int, + description: str, + message: str, + exception: Exception | None, + ) -> Status: + if code == 0: + raise ValueError( + "Custom error codes must not be zero since this value " + "is reserved for successful operations" + ) + elif code < 0: + raise ValueError("Custom error codes must not be smaller than zero") + + exception_type: type[Exception] | None = None + if exception is not None: + exception_type = type(exception) + + return Status( + code=code, + description=description, + message=message, + ExceptionType=exception_type, + ) + + def raise_for_status( + self, + state: Status, + ) -> None: + if state == self.SUCCESS: + if self.logger is not None: + self.logger.info( + "[STATUS] Raise for status - SUCCESS. all good.", stack_info=True + ) + return + + exc = self.error_to_exception(error_state=state) + raise exc + + +class NotSet: + __slots__ = tuple() + + def __init__(self) -> None: ... + + def __str__(self) -> str: + return ">Not set<" + + def __repr__(self) -> str: + return f"{self.__class__.__name__}()" + + +STATUS_HANDLER: t.Final[StatusHandler] = StatusHandler() + + +class ResultWrapper(t.Generic[T]): + __slots__ = ("_result", "status") + + def __init__( + self, + result: T | NotSet, + exception: Exception | None, + code_on_error: int, + ) -> None: + if isinstance(result, NotSet) and exception is None: + raise ValueError("[ResultWrapper] Set >NotSet< without exception") + elif not isinstance(result, NotSet) and exception is not None: + raise ValueError("[ResultWrapper] Set result with exception") + + 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 WrapperAccessResultDespiteError( + "Can not access result because it is not set" + ) + return self._result + + def __str__(self) -> str: + return f"Status(result: {self._result}, status: {self.status})" + + def __repr__(self) -> str: + return self.__str__() + + def unwrap(self) -> T: + STATUS_HANDLER.raise_for_status(self.status) + + return self.result + + +def wrap_result( + code_on_error: int, + logger: Logger | None = None, +) -> 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]: + wrapped_result: ResultWrapper[T] + try: + res = func(*args, **kwargs) + wrapped_result = ResultWrapper( + result=res, exception=None, code_on_error=code_on_error + ) + except Exception as err: + wrapped_result = ResultWrapper( + result=NotSet(), exception=err, code_on_error=code_on_error + ) + if logger is not None: + logger.info( + ( + "[RESULT-WRAPPER] An exception in routine %s occurred - msg: %s, " + "stack trace:" + ), + func.__name__, + str(err), + stack_info=True, + ) + + return wrapped_result + + return wrapper + + return wrap_result diff --git a/tests/test_result_pattern.py b/tests/test_result_pattern.py new file mode 100644 index 0000000..563e256 --- /dev/null +++ b/tests/test_result_pattern.py @@ -0,0 +1,268 @@ +import logging + +import pytest + +from dopt_basics import result_pattern + + +@pytest.fixture(scope="module") +def status_hdlr() -> result_pattern.StatusHandler: + return result_pattern.StatusHandler() + + +@pytest.fixture(scope="module") +def dummy_logger() -> logging.Logger: + logger = logging.getLogger("test") + handlers = tuple(logger.handlers) + for handler in handlers: + logger.removeHandler(handler) + logger.addHandler(logging.NullHandler()) + + return logger + + +def test_NotSet(): + not_set = result_pattern.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_status_handler_properties(status_hdlr, dummy_logger): + assert status_hdlr.logger is None + assert status_hdlr._SUCCESS is not None + assert status_hdlr.SUCCESS is not None + assert isinstance(status_hdlr.SUCCESS, result_pattern.Status) + assert status_hdlr.SUCCESS.code == 0 + assert status_hdlr.SUCCESS.description == "SUCCESS" + status_hdlr.logger = dummy_logger + assert status_hdlr.logger is not None + + +def test_status_handler_error_state_generation_FailCodeZero(status_hdlr): + description: str = "test" + message: str = "this is a test case" + code: int = 0 + with pytest.raises(ValueError, match="must not be zero since this value"): + status_hdlr.error_state(code, description, message, None) + + +def test_status_handler_error_state_generation_FailCodeNegative(status_hdlr): + description: str = "test" + message: str = "this is a test case" + code: int = -100 + with pytest.raises(ValueError, match="must not be smaller than zero"): + status_hdlr.error_state(code, description, message, None) + + +def test_status_handler_error_state_generation_SuccessWithoutException(status_hdlr): + description: str = "test" + message: str = "this is a test case" + code: int = 100 + err_state = status_hdlr.error_state(code, description, message, None) + assert err_state.code == code + assert err_state.description == description + assert err_state.message == message + assert err_state.ExceptionType is None + + +def test_status_handler_error_state_generation_SuccessWithException(status_hdlr): + description: str = "test" + message: str = "this is a test case" + code: int = 100 + exception: Exception = ValueError(message) + err_state = status_hdlr.error_state(code, description, message, exception) + assert err_state.code == code + assert err_state.description == description + assert err_state.message == message + assert err_state.ExceptionType is ValueError + + +def test_status_handler_error_to_exception_FailWithoutException(status_hdlr): + description: str = "test" + message: str = "this is a test case" + code: int = 100 + err_state = status_hdlr.error_state(code, description, message, None) + assert err_state.code == code + assert err_state.description == description + assert err_state.message == message + assert err_state.ExceptionType is None + with pytest.raises(ValueError, match="state where no exception is defined"): + status_hdlr.error_to_exception(err_state) + + +def test_status_handler_error_to_exception_SuccessWithException(status_hdlr): + description: str = "test" + message: str = "this is a test case" + code: int = 100 + exception: Exception = ValueError(message) + err_state = status_hdlr.error_state(code, description, message, exception) + assert err_state.code == code + assert err_state.description == description + assert err_state.message == message + assert err_state.ExceptionType is ValueError + exc = status_hdlr.error_to_exception(err_state) + assert isinstance(exc, ValueError) + assert str(exc) == message + + +def test_status_handler_exception_to_error_Success(status_hdlr): + description: str = "test" + message: str = "this is a test case" + code: int = 100 + + class TestException(Exception): + """test""" + + exc = TestException(message) + err_state = status_hdlr.exception_to_error(exc, code) + assert err_state.code == code + assert err_state.description == description + assert err_state.message == message + assert err_state.ExceptionType is TestException + + +def test_status_handler_raise_for_status_Success(dummy_logger): + status_hdlr = result_pattern.StatusHandler(dummy_logger) + state = status_hdlr.SUCCESS + + assert status_hdlr.raise_for_status(state) is None + + +def test_status_handler_raise_for_status_RaiseException(dummy_logger): + status_hdlr = result_pattern.StatusHandler(dummy_logger) + + description: str = "test" + message: str = "this is a test case" + code: int = 100 + exception: Exception = ValueError(message) + err_state = result_pattern.Status( + code=code, description=description, message=message, ExceptionType=type(exception) + ) + + with pytest.raises(type(exception), match=message): + status_hdlr.raise_for_status(err_state) + + +def test_result_wrapper_class_FailInitNotSetNoException(): + error_code = 146 + test_result = result_pattern.NotSet() + exception = None + + with pytest.raises(ValueError, match="Set >NotSet< without exception"): + _: result_pattern.ResultWrapper[int] = result_pattern.ResultWrapper( + result=test_result, + exception=exception, + code_on_error=error_code, + ) + + +def test_result_wrapper_class_FailInitSetResultWithException(): + error_code = 146 + test_result = 10 + message = "Test of error message" + exception = ValueError(message) + + with pytest.raises(ValueError, match="Set result with exception"): + _: result_pattern.ResultWrapper[int] = result_pattern.ResultWrapper( + result=test_result, + exception=exception, + code_on_error=error_code, + ) + + +def test_result_wrapper_class_Success(status_hdlr): + # successful operation + test_result = 10 + error_code = 146 + twrapper: result_pattern.ResultWrapper[int] = result_pattern.ResultWrapper( + result=test_result, + exception=None, + code_on_error=error_code, + ) + assert twrapper.status == status_hdlr.SUCCESS + assert twrapper.result == test_result + assert twrapper.unwrap() == test_result + assert twrapper.status.code != error_code + assert twrapper.status.code == status_hdlr.SUCCESS.code + + +def test_result_wrapper_class_FailureAccess(status_hdlr): + # test for no result + error_code = 146 + test_result = result_pattern.NotSet() + message = "Test of error message" + exception = ValueError(message) + twrapper: result_pattern.ResultWrapper[int] = result_pattern.ResultWrapper( + result=test_result, + exception=exception, + code_on_error=error_code, + ) + assert twrapper.status != status_hdlr.SUCCESS + assert twrapper.status.code == error_code + with pytest.raises(result_pattern.WrapperAccessResultDespiteError): + twrapper.result + with pytest.raises(type(exception), match=message): + _ = twrapper.unwrap() + + +def test_wrap_result_ExceptionRaised(): + MESSAGE = "Test case wrapped function decorator" + error_code = 103 + + @result_pattern.wrap_result(error_code) + def test_func_1() -> None: + raise ValueError(MESSAGE) + + res = test_func_1() + assert isinstance(res, result_pattern.ResultWrapper) + assert res.status.code == error_code + assert res.status.message == MESSAGE + assert res.status.ExceptionType is ValueError + with pytest.raises(ValueError, match=MESSAGE): + res.unwrap() + + +def test_wrap_result_NotNone_Success(status_hdlr): + error_code = 103 + + @result_pattern.wrap_result(error_code) + def test_func_2(x: str, y: str) -> int: + return int(int(x) / int(y)) + + res = test_func_2("2", "1") + assert res.result == 2 + assert res.unwrap() == 2 + assert res.status == status_hdlr.SUCCESS + + +def test_wrap_result_NotNone_FailureZeroDivisionError(): + error_code = 103 + + @result_pattern.wrap_result(error_code) + def test_func_2(x: str, y: str) -> int: + return int(int(x) / int(y)) + + res = test_func_2("2", "0") + with pytest.raises(result_pattern.WrapperAccessResultDespiteError): + res.result + with pytest.raises(ZeroDivisionError): + res.unwrap() + assert res.status.code == error_code + assert res.status.ExceptionType is ZeroDivisionError + + +def test_wrap_result_NotNone_FailureValueErrorWithLogger(dummy_logger): + error_code = 103 + + @result_pattern.wrap_result(error_code, logger=dummy_logger) + def test_func_2(x: str, y: str) -> int: + return int(int(x) / int(y)) + + res = test_func_2("2", "test") + with pytest.raises(result_pattern.WrapperAccessResultDespiteError): + res.result + with pytest.raises(ValueError): + res.unwrap() + assert res.status.code == error_code + assert res.status.ExceptionType is ValueError