pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "dopt-basics"
-version = "0.2.4"
+version = "0.1.3"
 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.2.4"
+current_version = "0.1.3"
 parse = """(?x)
     (?P<major>0|[1-9]\\d*)\\.
     (?P<minor>0|[1-9]\\d*)\\.

scripts/bump_patch.ps1

@@ -1,2 +0,0 @@
-pdm run bump-my-version bump patch
-pdm run bump-my-version show current_version

src/dopt_basics/cli.py

@@ -1,108 +0,0 @@
-import shutil
-import sys
-import time
-import typing as t
-from collections.abc import Callable
-from functools import wraps
-from itertools import cycle
-from threading import Thread
-
-P = t.ParamSpec("P")
-T = t.TypeVar("T")
-
-
-# based on: https://stackoverflow.com/questions/22029562/python-how-to-make-simple-animated-loading-while-process-is-running
-class LoadingAnimation:
-    def __init__(
-        self,
-        loading_txt: str,
-        ending_text: str,
-        timeout: float = 0.1,
-        force_ascii: bool = False,
-    ) -> None:
-        self.loading_txt = loading_txt
-        self.ending_text = ending_text
-        self.timeout = timeout
-        self.done: bool = False
-        self.keyboard_interrupt: bool = False
-
-        self._isatty = sys.stdout.isatty()
-        self._do_animation = bool(sys.stdout.isatty())
-
-        self._animate_dots: list[str] = ["⢿", "⣻", "⣽", "⣾", "⣷", "⣯", "⣟", "⡿"]
-        self._animate_ascii: list[str] = ["|", "/", "-", "\\"]
-
-        self.frames: list[str]
-        enc = (sys.stdout.encoding or "utf-8").lower()
-        if force_ascii:
-            self.frames = self._animate_ascii
-        else:
-            try:
-                self._animate_dots[0].encode(enc)
-                self.frames = self._animate_dots
-            except Exception:
-                self.frames = self._animate_ascii
-
-        self._thread: Thread = Thread(target=self._animation)
-
-    def __enter__(self) -> t.Self:
-        self.start()
-        return self
-
-    def __exit__(
-        self,
-        exc_type: type[Exception],
-        exc_value,
-        tb,
-    ) -> None:
-        if exc_type is not None:
-            self.stop(interrupt=True)
-            if exc_type is KeyboardInterrupt:
-                self.keyboard_interrupt = True
-                print("Operation cancelled by user. (KeyboardInterrupt)", flush=True)
-            return
-
-        self.stop()
-
-    def _animation(self) -> None:
-        for frame in cycle(self.frames):
-            if self.done:
-                break
-            print(f"\r{frame} {self.loading_txt}", flush=True, end="")
-            time.sleep(self.timeout)
-
-    def start(self) -> None:
-        if self._do_animation:
-            self._thread.start()
-        else:
-            print(f"\r{self.loading_txt}", end="", flush=True)
-
-    def stop(
-        self,
-        interrupt: bool = False,
-    ) -> None:
-        if self.done:
-            return
-
-        self.done = True
-        if self._do_animation:
-            self._thread.join()
-        cols = shutil.get_terminal_size((80, 20)).columns
-        print("\r" + " " * cols, end="\r", flush=True)
-        if interrupt:
-            return
-        print(f"{self.ending_text}", flush=True)
-
-
-def default_loading(func: Callable[P, T]) -> Callable[P, T]:
-    @wraps(func)
-    def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
-        loading_txt: str = "Performing operation..."
-        ending_text: str = "Operation successful"
-
-        with LoadingAnimation(loading_txt, ending_text):
-            res = func(*args, **kwargs)
-
-        return res
-
-    return wrapper

src/dopt_basics/result_pattern.py

@@ -1,201 +0,0 @@
-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

tests/test_result_pattern.py

@@ -1,268 +0,0 @@
-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