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