From 3f9d5e36529c67718016e3305aabc9fb2ea2b2e3 Mon Sep 17 00:00:00 2001 From: foefl Date: Wed, 22 Oct 2025 16:52:46 +0200 Subject: [PATCH] basic structure for generic result pattern --- src/dopt_basics/result_pattern.py | 205 ++++++++++++++++++++++++++++++ 1 file changed, 205 insertions(+) create mode 100644 src/dopt_basics/result_pattern.py diff --git a/src/dopt_basics/result_pattern.py b/src/dopt_basics/result_pattern.py new file mode 100644 index 0000000..2c52da1 --- /dev/null +++ b/src/dopt_basics/result_pattern.py @@ -0,0 +1,205 @@ +import dataclasses as dc +import inspect +import typing as t +from collections.abc import Callable +from functools import wraps +from logging import Logger + +P = t.ParamSpec("P") +T = t.TypeVar("T") + + +# ** 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)""" + + +class UInternalError(Exception): + """unwrap exception: all internal errors; internal, if error is not caused by (external) + data related issues""" + + +class UApiError(Exception): + """unwrap exception: all errors occurred on the server side of Delta Barth's API + default case: should not be raised, as this kind of errors should be handled by + Delta Barth themselves""" + + +@dc.dataclass(kw_only=True, slots=True) +class Status: + code: int + description: str + message: str + ExceptionType: type[Exception] | None = None + + +class StatusHandler: + 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]): + def __init__( + self, + result: T | NotSet, + exception: Exception | None, + code_on_error: int, + ) -> None: + assert (isinstance(result, NotSet) and exception is not None) or ( + not isinstance(result, NotSet) and exception is None + ), "set >NotSet< without exception or 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 WAccessResultDespiteError("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