Fixed spinner thread & remove log thread

src/cpl-core/cpl/core/configuration/configuration.py

@@ -115,18 +115,7 @@ class Configuration:
                 if sub.__name__ != key and sub.__name__.replace("Settings", "") != key:
                     continue
 
-                configuration = sub()
+                configuration = JSONProcessor.process(sub, value)
-                from_dict = getattr(configuration, "from_dict", None)
-
-                if from_dict is not None and not hasattr(from_dict, "is_base_func"):
-                    Console.set_foreground_color(ForegroundColorEnum.yellow)
-                    Console.write_line(
-                        f"{sub.__name__}.from_dict is deprecated. Instead, set attributes as typed arguments in __init__. They can be None by default!"
-                    )
-                    Console.color_reset()
-                    configuration.from_dict(value)
-                else:
-                    configuration = JSONProcessor.process(sub, value)
 
                 cls.set(sub, configuration)
 

src/cpl-core/cpl/core/configuration/configuration_model_abc.py

@@ -1,21 +1,5 @@
-from abc import ABC, abstractmethod
+from abc import ABC
-
-
-def base_func(method):
-    method.is_base_func = True
-    return method
 
 
 class ConfigurationModelABC(ABC):
-    @abstractmethod
+    pass
-    def __init__(self):
-        r"""ABC for settings representation"""
-
-    @base_func
-    def from_dict(self, settings: dict):
-        r"""DEPRECATED: Set attributes as typed arguments in __init__ instead. See https://docs.sh-edraft.de/cpl/deprecated.html#ConfigurationModelABC-from_dict-method for further information
-        Converts attributes to dict
-
-        Parameter:
-            settings: :class:`dict`
-        """

src/cpl-core/cpl/core/console/__init__.py

@@ -1,5 +1,4 @@
 from .background_color_enum import BackgroundColorEnum
 from .console import Console
-from .console_call import ConsoleCall
+from ._call import ConsoleCall
 from .foreground_color_enum import ForegroundColorEnum
-from .spinner_thread import SpinnerThread

src/cpl-core/cpl/core/console/_spinner.py

@@ -1,7 +1,8 @@
 import os
 import sys
-import threading
+import multiprocessing
 import time
+from multiprocessing import Process
 
 from termcolor import colored
 
@@ -9,8 +10,8 @@ from cpl.core.console.background_color_enum import BackgroundColorEnum
 from cpl.core.console.foreground_color_enum import ForegroundColorEnum
 
 
-class SpinnerThread(threading.Thread):
+class Spinner(Process):
-    r"""Thread to show spinner in terminal
+    r"""Process to show spinner in terminal
 
     Parameter:
         msg_len: :class:`int`
@@ -22,7 +23,7 @@ class SpinnerThread(threading.Thread):
     """
 
     def __init__(self, msg_len: int, foreground_color: ForegroundColorEnum, background_color: BackgroundColorEnum):
-        threading.Thread.__init__(self)
+        Process.__init__(self)
 
         self._msg_len = msg_len
         self._foreground_color = foreground_color
@@ -50,29 +51,26 @@ class SpinnerThread(threading.Thread):
         return color_args
 
     def run(self) -> None:
-        r"""Entry point of thread, shows the spinner"""
+        r"""Entry point of process, shows the spinner"""
         columns = 0
         if sys.platform == "win32":
             columns = os.get_terminal_size().columns
         else:
-            term_rows, term_columns = os.popen("stty size", "r").read().split()
+            values = os.popen("stty size", "r").read().split()
+            term_rows, term_columns = values if len(values) == 2 else (0, 0)
             columns = int(term_columns)
 
         end_msg = "done"
-        end_msg_pos = columns - self._msg_len - len(end_msg)
+
-        if end_msg_pos > 0:
+        padding = columns - self._msg_len - len(end_msg)
-            print(f'{"" : >{end_msg_pos}}', end="")
+        if padding > 0:
+            print(f'{"" : >{padding}}', end="")
         else:
             print("", end="")
 
-        first = True
         spinner = self._spinner()
         while self._is_spinning:
-            if first:
+            print(colored(f"{next(spinner): >{len(end_msg)}}", *self._get_color_args()), end="")
-                first = False
-                print(colored(f"{next(spinner): >{len(end_msg) - 1}}", *self._get_color_args()), end="")
-            else:
-                print(colored(f"{next(spinner): >{len(end_msg)}}", *self._get_color_args()), end="")
             time.sleep(0.1)
             back = ""
             for i in range(0, len(end_msg)):
@@ -84,9 +82,10 @@ class SpinnerThread(threading.Thread):
         if not self._exit:
             print(colored(end_msg, *self._get_color_args()), end="")
 
-    def stop_spinning(self):
+    def stop(self):
         r"""Stops the spinner"""
         self._is_spinning = False
+        super().terminate()
         time.sleep(0.1)
 
     def exit(self):

src/cpl-core/cpl/core/console/console.py

@@ -10,9 +10,9 @@ from tabulate import tabulate
 from termcolor import colored
 
 from cpl.core.console.background_color_enum import BackgroundColorEnum
-from cpl.core.console.console_call import ConsoleCall
+from cpl.core.console._call import ConsoleCall
 from cpl.core.console.foreground_color_enum import ForegroundColorEnum
-from cpl.core.console.spinner_thread import SpinnerThread
+from cpl.core.console._spinner import Spinner
 
 
 class Console:
@@ -464,7 +464,7 @@ class Console:
         cls.set_hold_back(True)
         spinner = None
         if not cls._disabled:
-            spinner = SpinnerThread(len(message), spinner_foreground_color, spinner_background_color)
+            spinner = Spinner(len(message), spinner_foreground_color, spinner_background_color)
             spinner.start()
 
         return_value = None
@@ -476,7 +476,7 @@ class Console:
             cls.close()
 
         if spinner is not None:
-            spinner.stop_spinning()
+            spinner.stop()
         cls.set_hold_back(False)
 
         cls.set_foreground_color(ForegroundColorEnum.default)

src/cpl-core/cpl/core/log/_log_writer.py

@@ -1,92 +0,0 @@
-import multiprocessing
-import os
-from datetime import datetime
-from typing import Self
-
-from cpl.core.console import Console
-from cpl.core.log.log_level_enum import LogLevelEnum
-
-
-class LogWriter:
-    _instance = None
-
-    # ANSI color codes for different log levels
-    _COLORS = {
-        LogLevelEnum.trace: "\033[37m",  # Light Gray
-        LogLevelEnum.debug: "\033[94m",  # Blue
-        LogLevelEnum.info: "\033[92m",  # Green
-        LogLevelEnum.warning: "\033[93m",  # Yellow
-        LogLevelEnum.error: "\033[91m",  # Red
-        LogLevelEnum.fatal: "\033[95m",  # Magenta
-    }
-
-    def __init__(self, file_prefix: str, level: LogLevelEnum = LogLevelEnum.info):
-        self._file_prefix = file_prefix
-        self._level = level
-
-        self._queue = multiprocessing.Queue()
-        self._process = multiprocessing.Process(target=self._log_worker, daemon=True)
-
-        self._create_log_dir()
-        self._process.start()
-
-    @property
-    def level(self) -> LogLevelEnum:
-        return self._level
-
-    @level.setter
-    def level(self, value: LogLevelEnum):
-        assert isinstance(value, LogLevelEnum), "Log level must be an instance of LogLevelEnum"
-        self._level = value
-
-    @classmethod
-    def get_instance(cls, file_prefix: str, level: LogLevelEnum = LogLevelEnum.info) -> Self:
-        if cls._instance is None:
-            cls._instance = LogWriter(file_prefix, level)
-        return cls._instance
-
-    @staticmethod
-    def _create_log_dir():
-        if os.path.exists("logs"):
-            return
-
-        os.makedirs("logs")
-
-    def _log_worker(self):
-        """Worker process that writes log messages from the queue to the file."""
-        while True:
-            content = self._queue.get()
-            if content is None:  # Shutdown signal
-                break
-            self._write_log_to_file(content)
-            Console.write_line(f"{self._COLORS.get(self._level, '\033[0m')}{content}\033[0m")
-
-    @property
-    def log_file(self):
-        return f"logs/{self._file_prefix}_{datetime.now().strftime('%Y-%m-%d')}.log"
-
-    def _ensure_file_size(self):
-        log_file = self.log_file
-        if not os.path.exists(log_file) or os.path.getsize(log_file) <= 0.5 * 1024 * 1024:
-            return
-
-        # if exists and size is greater than 300MB, create a new file
-        os.rename(
-            log_file,
-            f"{log_file.split('.log')[0]}_{datetime.now().strftime('%H-%M-%S')}.log",
-        )
-
-    def _write_log_to_file(self, content: str):
-        self._ensure_file_size()
-        with open(self.log_file, "a") as log_file:
-            log_file.write(content + "\n")
-            log_file.close()
-
-    def log(self, content: str):
-        """Enqueue log message without blocking main app."""
-        self._queue.put(content)
-
-    def close(self):
-        """Gracefully stop the logging process."""
-        self._queue.put(None)
-        self._process.join()

src/cpl-core/cpl/core/log/logger.py

@@ -3,7 +3,6 @@ import traceback
 from datetime import datetime
 
 from cpl.core.console import Console
-from cpl.core.log._log_writer import LogWriter
 from cpl.core.log.log_level_enum import LogLevelEnum
 from cpl.core.log.logger_abc import LoggerABC
 from cpl.core.typing import Messages, Source
@@ -13,6 +12,16 @@ class Logger(LoggerABC):
     _level = LogLevelEnum.info
     _levels = [x for x in LogLevelEnum]
 
+    # ANSI color codes for different log levels
+    _COLORS = {
+        LogLevelEnum.trace: "\033[37m",  # Light Gray
+        LogLevelEnum.debug: "\033[94m",  # Blue
+        LogLevelEnum.info: "\033[92m",  # Green
+        LogLevelEnum.warning: "\033[93m",  # Yellow
+        LogLevelEnum.error: "\033[91m",  # Red
+        LogLevelEnum.fatal: "\033[95m",  # Magenta
+    }
+
     def __init__(self, source: Source, file_prefix: str = None):
         LoggerABC.__init__(self)
         assert source is not None and source != "", "Source cannot be None or empty"
@@ -22,7 +31,18 @@ class Logger(LoggerABC):
             file_prefix = "app"
 
         self._file_prefix = file_prefix
-        self._writer = LogWriter.get_instance(self._file_prefix)
+        self._create_log_dir()
+
+    @property
+    def log_file(self):
+        return f"logs/{self._file_prefix}_{datetime.now().strftime('%Y-%m-%d')}.log"
+
+    @staticmethod
+    def _create_log_dir():
+        if os.path.exists("logs"):
+            return
+
+        os.makedirs("logs")
 
     @classmethod
     def set_level(cls, level: LogLevelEnum):
@@ -31,6 +51,24 @@ class Logger(LoggerABC):
         else:
             raise ValueError(f"Invalid log level: {level}")
 
+    @staticmethod
+    def _ensure_file_size(log_file: str):
+        if not os.path.exists(log_file) or os.path.getsize(log_file) <= 0.5 * 1024 * 1024:
+            return
+
+        # if exists and size is greater than 300MB, create a new file
+        os.rename(
+            log_file,
+            f"{log_file.split('.log')[0]}_{datetime.now().strftime('%H-%M-%S')}.log",
+        )
+
+    def _write_log_to_file(self, content: str):
+        file = self.log_file
+        self._ensure_file_size(file)
+        with open(file, "a") as log_file:
+            log_file.write(content + "\n")
+            log_file.close()
+
     def _log(self, level: LogLevelEnum, *messages: Messages):
         try:
             if self._levels.index(level) < self._levels.index(self._level):
@@ -39,7 +77,8 @@ class Logger(LoggerABC):
             timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")
             formatted_message = self._format_message(level.value, timestamp, *messages)
 
-            self._writer.log(formatted_message)
+            self._write_log_to_file(formatted_message)
+            Console.write_line(f"{self._COLORS.get(self._level, '\033[0m')}{formatted_message}\033[0m")
         except Exception as e:
             print(f"Error while logging: {e} -> {traceback.format_exc()}")
 

tests/custom/console/main.py

@@ -25,6 +25,7 @@ if __name__ == "__main__":
     Console.spinner(
         "Test:", test_spinner, spinner_foreground_color=ForegroundColorEnum.cyan, text_foreground_color="green"
     )
+    Console.write_line("HOLD BACK")
     # opts = [
     #     'Option 1',
     #     'Option 2',

tests/custom/database/src/application.py

@@ -1,13 +1,12 @@
 from typing import Optional
 
 from cpl.application import ApplicationABC
-from cpl.core.configuration import Configuration
 from cpl.core.console import Console
 from cpl.core.environment import Environment
-from cpl.dependency import ServiceProviderABC
 from cpl.core.log import LoggerABC
-from model.user_repo_abc import UserRepoABC
+from cpl.dependency import ServiceProviderABC
 from model.user_repo import UserRepo
+from model.user_repo_abc import UserRepoABC
 
 
 class Application(ApplicationABC):