allow early stop calls to CLI spinner

pyproject.toml

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

src/dopt_basics/cli.py

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