Bug fix for logging queue listener and ignoring of per handler set levels

in the QueueLister launcher add "respect_handler_level=True" so it respects the previous set log levels per handler

Also split all logging tests into their own file

pyproject.toml

@@ -1,7 +1,7 @@
 # MARK: Project info
 [project]
 name = "corelibs"
-version = "0.9.0"
+version = "0.10.0"
 description = "Collection of utils for Python scripts"
 readme = "README.md"
 requires-python = ">=3.13"

src/corelibs/logging_handling/log.py

@@ -43,7 +43,6 @@ class CustomConsoleFormatter(logging.Formatter):
         "ERROR": Colors.red,
         "CRITICAL": Colors.magenta,
         "EXCEPTION": Colors.magenta_bold,
-        "RESET": Colors.reset,  # Reset
     }
 
     def format(self, record: logging.LogRecord) -> str:
@@ -57,7 +56,7 @@ class CustomConsoleFormatter(logging.Formatter):
             str -- _description_
         """
         # Add color to levelname for console output
-        reset = self.COLORS["RESET"]
+        reset = Colors.reset
         color = self.COLORS.get(record.levelname, reset)
         # only highlight level for basic
         if record.levelname in ['DEBUG', 'INFO']:
@@ -154,17 +153,17 @@ class Log:
     ) -> LogSettings:
         # skip with defaul it not set
         if log_settings is None:
-            return __class__.DEFAULT_LOG_SETTINGS
+            return self.DEFAULT_LOG_SETTINGS
         # check entries
-        default_log_settings = __class__.DEFAULT_LOG_SETTINGS
+        default_log_settings = self.DEFAULT_LOG_SETTINGS
         # check log levels
         for __log_entry in ['log_level_console', 'log_level_file']:
             if log_settings.get(__log_entry) is None:
                 continue
             # if not valid reset to default, if not in default set to WARNING
             if not self.validate_log_level(_log_level := log_settings.get(__log_entry, '')):
-                _log_level = __class__.DEFAULT_LOG_SETTINGS.get(
-                    __log_entry, __class__.DEFAULT_LOG_LEVEL
+                _log_level = self.DEFAULT_LOG_SETTINGS.get(
+                    __log_entry, self.DEFAULT_LOG_LEVEL
                 )
             default_log_settings[__log_entry] = str(_log_level)
         # check bool
@@ -177,10 +176,10 @@ class Log:
             if log_settings.get(__log_entry) is None:
                 continue
             if not isinstance(__setting := log_settings.get(__log_entry, ''), bool):
-                __setting = __class__.DEFAULT_LOG_SETTINGS.get(__log_entry, True)
+                __setting = self.DEFAULT_LOG_SETTINGS.get(__log_entry, True)
             default_log_settings[__log_entry] = __setting
         # check log queue
-        __setting = log_settings.get('log_queue', __class__.DEFAULT_LOG_SETTINGS['log_queue'])
+        __setting = log_settings.get('log_queue', self.DEFAULT_LOG_SETTINGS['log_queue'])
         if __setting is not None:
             __setting = cast('Queue[str]', __setting)
         default_log_settings['log_queue'] = __setting
@@ -255,7 +254,11 @@ class Log:
         if log_queue is None:
             return
         self.log_queue = log_queue
-        self.listener = logging.handlers.QueueListener(self.log_queue, *self.handlers)
+        self.listener = logging.handlers.QueueListener(
+            self.log_queue,
+            *self.handlers,
+            respect_handler_level=True
+        )
         self.listener.start()
 
     def __init_log(self, log_name: str) -> None:
@@ -308,7 +311,7 @@ class Log:
         Keyword Arguments:
             info {str} -- _description_ (default: {"BREAK"})
         """
-        self.logger.info("[%s] %s>", info, __class__.SPACER_CHAR * __class__.SPACER_LENGTH)
+        self.logger.info("[%s] %s>", info, self.SPACER_CHAR * self.SPACER_LENGTH)
 
     def exception(self, msg: object, *args: object, extra: Mapping[str, object] | None = None) -> None:
         """

src/corelibs/string_handling/text_colors.py

@@ -10,40 +10,73 @@ class Colors:
     """
     ANSI colors defined
     """
-    # General sets
-    bold = '\033[1m'
-    underline = '\033[4m'
-    end = '\033[0m'
-    reset = '\033[0m'
+    # General sets, these should not be accessd
+    __BOLD = '\033[1m'
+    __UNDERLINE = '\033[4m'
+    __END = '\033[0m'
+    __RESET = '\033[0m'
     # Define ANSI color codes as class attributes
-    black = "\033[30m"
-    red = "\033[31m"
-    green = "\033[32m"
-    yellow = "\033[33m"
-    blue = "\033[34m"
-    magenta = "\033[35m"
-    cyan = "\033[36m"
-    white = "\033[37m"
+    __BLACK = "\033[30m"
+    __RED = "\033[31m"
+    __GREEN = "\033[32m"
+    __YELLOW = "\033[33m"
+    __BLUE = "\033[34m"
+    __MAGENTA = "\033[35m"
+    __CYAN = "\033[36m"
+    __WHITE = "\033[37m"
 
     # Define bold/bright versions of the colors
-    black_bold = "\033[1;30m"
-    red_bold = "\033[1;31m"
-    green_bold = "\033[1;32m"
-    yellow_bold = "\033[1;33m"
-    blue_bold = "\033[1;34m"
-    magenta_bold = "\033[1;35m"
-    cyan_bold = "\033[1;36m"
-    white_bold = "\033[1;37m"
+    __BLACK_BOLD = "\033[1;30m"
+    __RED_BOLD = "\033[1;31m"
+    __GREEN_BOLD = "\033[1;32m"
+    __YELLOW_BOLD = "\033[1;33m"
+    __BLUE_BOLD = "\033[1;34m"
+    __MAGENTA_BOLD = "\033[1;35m"
+    __CYAN_BOLD = "\033[1;36m"
+    __WHITE_BOLD = "\033[1;37m"
 
     # BRIGHT, alternative
-    black_bright = '\033[90m'
-    red_bright = '\033[91m'
-    green_bright = '\033[92m'
-    yellow_bright = '\033[93m'
-    blue_bright = '\033[94m'
-    magenta_bright = '\033[95m'
-    cyan_bright = '\033[96m'
-    white_bright = '\033[97m'
+    __BLACK_BRIGHT = '\033[90m'
+    __RED_BRIGHT = '\033[91m'
+    __GREEN_BRIGHT = '\033[92m'
+    __YELLOW_BRIGHT = '\033[93m'
+    __BLUE_BRIGHT = '\033[94m'
+    __MAGENTA_BRIGHT = '\033[95m'
+    __CYAN_BRIGHT = '\033[96m'
+    __WHITE_BRIGHT = '\033[97m'
+
+    # set access vars
+    bold = __BOLD
+    underline = __UNDERLINE
+    end = __END
+    reset = __RESET
+    # normal
+    black = __BLACK
+    red = __RED
+    green = __GREEN
+    yellow = __YELLOW
+    blue = __BLUE
+    magenta = __MAGENTA
+    cyan = __CYAN
+    white = __WHITE
+    # bold
+    black_bold = __BLACK_BOLD
+    red_bold = __RED_BOLD
+    green_bold = __GREEN_BOLD
+    yellow_bold = __YELLOW_BOLD
+    blue_bold = __BLUE_BOLD
+    magenta_bold = __MAGENTA_BOLD
+    cyan_bold = __CYAN_BOLD
+    white_bold = __WHITE_BOLD
+    # bright
+    black_bright = __BLACK_BRIGHT
+    red_bright = __RED_BRIGHT
+    green_bright = __GREEN_BRIGHT
+    yellow_bright = __YELLOW_BRIGHT
+    blue_bright = __BLUE_BRIGHT
+    magenta_bright = __MAGENTA_BRIGHT
+    cyan_bright = __CYAN_BRIGHT
+    white_bright = __WHITE_BRIGHT
 
     @staticmethod
     def disable():
@@ -81,4 +114,43 @@ class Colors:
         Colors.magenta_bright = ''
         Colors.cyan_bright = ''
         Colors.white_bright = ''
+
+    @staticmethod
+    def reset_colors():
+        """
+        reset colors to the original ones
+        """
+        # set access vars
+        Colors.bold = Colors.__BOLD
+        Colors.underline = Colors.__UNDERLINE
+        Colors.end = Colors.__END
+        Colors.reset = Colors.__RESET
+        # normal
+        Colors.black = Colors.__BLACK
+        Colors.red = Colors.__RED
+        Colors.green = Colors.__GREEN
+        Colors.yellow = Colors.__YELLOW
+        Colors.blue = Colors.__BLUE
+        Colors.magenta = Colors.__MAGENTA
+        Colors.cyan = Colors.__CYAN
+        Colors.white = Colors.__WHITE
+        # bold
+        Colors.black_bold = Colors.__BLACK_BOLD
+        Colors.red_bold = Colors.__RED_BOLD
+        Colors.green_bold = Colors.__GREEN_BOLD
+        Colors.yellow_bold = Colors.__YELLOW_BOLD
+        Colors.blue_bold = Colors.__BLUE_BOLD
+        Colors.magenta_bold = Colors.__MAGENTA_BOLD
+        Colors.cyan_bold = Colors.__CYAN_BOLD
+        Colors.white_bold = Colors.__WHITE_BOLD
+        # bright
+        Colors.black_bright = Colors.__BLACK_BRIGHT
+        Colors.red_bright = Colors.__RED_BRIGHT
+        Colors.green_bright = Colors.__GREEN_BRIGHT
+        Colors.yellow_bright = Colors.__YELLOW_BRIGHT
+        Colors.blue_bright = Colors.__BLUE_BRIGHT
+        Colors.magenta_bright = Colors.__MAGENTA_BRIGHT
+        Colors.cyan_bright = Colors.__CYAN_BRIGHT
+        Colors.white_bright = Colors.__WHITE_BRIGHT
+
 # __END__

src/corelibs/string_handling/timestamp_strings.py

@@ -11,11 +11,11 @@ class TimestampStrings:
     set default time stamps
     """
 
-    time_zone: str = 'Asia/Tokyo'
+    TIME_ZONE: str = 'Asia/Tokyo'
 
     def __init__(self, time_zone: str | None = None):
         self.timestamp_now = datetime.now()
-        self.time_zone = time_zone if time_zone is not None else __class__.time_zone
+        self.time_zone = time_zone if time_zone is not None else self.TIME_ZONE
         try:
             self.timestamp_now_tz = datetime.now(ZoneInfo(self.time_zone))
         except ZoneInfoNotFoundError as e:

test-run/logging_handling/log.py

@@ -4,9 +4,7 @@ Log logging_handling.log testing
 
 # import atexit
 from pathlib import Path
-from multiprocessing import Queue
 # this is for testing only
-from queue_logger.log_queue import QueueLogger
 from corelibs.logging_handling.log import Log
 
 
@@ -19,27 +17,18 @@ def main():
         log_path=script_path.joinpath('log', 'test.log'),
         log_name="Test Log",
         log_settings={
-            "log_level_console": 'DEBUG',
+            "log_level_console": 'WARNING',
             "log_level_file": 'DEBUG',
             # "console_color_output_enabled": False,
         }
     )
 
-    log.logger.debug('Debug test: %s', log.logger.name)
-    log.logger.info('Info test: %s', log.logger.name)
-    log.logger.warning('Warning test: %s', log.logger.name)
-    log.logger.error('Error test: %s', log.logger.name)
-    log.logger.critical('Critical test: %s', log.logger.name)
-    log.exception('Exception test: %s', log.logger.name)
-
-    log_queue: 'Queue[str]' = Queue()
-    log_q = QueueLogger(
-        log_file=script_path.joinpath('log', 'test_queue.log'),
-        log_name="Test Log Queue",
-        log_queue=log_queue
-    )
-    log_q.mlog.info('Log test: %s', log.logger.name)
-    # log_q.stop_listener()
+    log.logger.debug('[NORMAL] Debug test: %s', log.logger.name)
+    log.logger.info('[NORMAL] Info test: %s', log.logger.name)
+    log.logger.warning('[NORMAL] Warning test: %s', log.logger.name)
+    log.logger.error('[NORMAL] Error test: %s', log.logger.name)
+    log.logger.critical('[NORMAL] Critical test: %s', log.logger.name)
+    log.exception('[NORMAL] Exception test: %s', log.logger.name)
 
 
 if __name__ == "__main__":

test-run/logging_handling/log_pool.py

@@ -0,0 +1,75 @@
+"""
+Pool Queue log handling
+Thread Queue log handling
+"""
+
+import random
+import time
+from multiprocessing import Queue
+import concurrent.futures
+import logging
+from pathlib import Path
+from corelibs.logging_handling.log import Log
+
+
+def work_function(log_name: str, worker_id: int, data: list[int]) -> int:
+    """
+    simulate worker
+
+    Arguments:
+        worker_id {int} -- _description_
+        data {list[int]} -- _description_
+
+    Returns:
+        int -- _description_
+    """
+    log = logging.getLogger(f'{log_name}-WorkerFn-{worker_id}')
+    log.info('Starting worker: %s', worker_id)
+    time.sleep(random.uniform(1, 3))
+    result = sum(data) * worker_id
+    return result
+
+
+def main():
+    """
+    Queue log tester
+    """
+    print("[START] Queue logger test")
+    log_queue: 'Queue[str]' = Queue()
+    script_path: Path = Path(__file__).resolve().parent
+    log = Log(
+        log_path=script_path.joinpath('log', 'test.log'),
+        log_name="Test Log",
+        log_settings={
+            "log_level_console": 'INFO',
+            "log_level_file": 'INFO',
+            "log_queue": log_queue,
+        }
+    )
+    log.logger.info('Pool Fork logging test')
+    max_forks = 2
+    data_sets = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]
+    with concurrent.futures.ProcessPoolExecutor(
+        max_workers=max_forks,
+        initializer=Log.init_worker_logging,
+        initargs=(log_queue,)
+    ) as executor:
+        log.logger.info('Start workers')
+        futures = [
+            executor.submit(work_function, log.log_name, worker_id, data)
+            for worker_id, data in enumerate(data_sets, 1)
+        ]
+        log.logger.info('Workders started')
+
+        for future in concurrent.futures.as_completed(futures):
+            log.logger.info('Processing result: %s', future.result())
+            print(f"Processing result: {future.result()}")
+
+    log.logger.info('[END] Queue logger test')
+    log.stop_listener()
+
+
+if __name__ == "__main__":
+    main()
+
+# __END__

test-run/logging_handling/log_queue.py

@@ -1,72 +1,39 @@
 """
-Pool Queue log handling
-Thread Queue log handling
+Log logging_handling.log testing
 """
 
-import random
-import time
-from multiprocessing import Queue
-import concurrent.futures
-import logging
+# import atexit
 from pathlib import Path
+from multiprocessing import Queue
+# this is for testing only
 from corelibs.logging_handling.log import Log
 
 
-def work_function(log_name: str, worker_id: int, data: list[int]) -> int:
-    """
-    simulate worker
-
-    Arguments:
-        worker_id {int} -- _description_
-        data {list[int]} -- _description_
-
-    Returns:
-        int -- _description_
-    """
-    log = logging.getLogger(f'{log_name}-WorkerFn-{worker_id}')
-    log.info('Starting worker: %s', worker_id)
-    time.sleep(random.uniform(1, 3))
-    result = sum(data) * worker_id
-    return result
-
-
 def main():
     """
-    Queue log tester
+    Log testing
     """
-    print("[START] Queue logger test")
-    log_queue: 'Queue[str]' = Queue()
     script_path: Path = Path(__file__).resolve().parent
-    log = Log(
-        log_path=script_path.joinpath('log', 'test.log'),
+
+    log_queue: 'Queue[str]' = Queue()
+    log_q = Log(
+        log_path=script_path.joinpath('log', 'test_queue.log'),
         log_name="Test Log",
         log_settings={
-            "log_level_console": 'INFO',
-            "log_level_file": 'INFO',
-            "log_queue": log_queue,
+            "log_level_console": 'WARNING',
+            "log_level_file": 'ERROR',
+            "log_queue": log_queue
+            # "console_color_output_enabled": False,
         }
     )
-    log.logger.info('Pool Fork logging test')
-    max_forks = 2
-    data_sets = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]
-    with concurrent.futures.ProcessPoolExecutor(
-        max_workers=max_forks,
-        initializer=Log.init_worker_logging,
-        initargs=(log_queue,)
-    ) as executor:
-        log.logger.info('Start workers')
-        futures = [
-            executor.submit(work_function, log.log_name, worker_id, data)
-            for worker_id, data in enumerate(data_sets, 1)
-        ]
-        log.logger.info('Workders started')
 
-        for future in concurrent.futures.as_completed(futures):
-            log.logger.info('Processing result: %s', future.result())
-            print(f"Processing result: {future.result()}")
-
-    log.logger.info('[END] Queue logger test')
-    log.stop_listener()
+    log_q.logger.debug('[QUEUE] Debug test: %s', log_q.logger.name)
+    log_q.logger.info('[QUEUE] Info test: %s', log_q.logger.name)
+    log_q.logger.warning('[QUEUE] Warning test: %s', log_q.logger.name)
+    log_q.logger.error('[QUEUE] Error test: %s', log_q.logger.name)
+    log_q.logger.critical('[QUEUE] Critical test: %s', log_q.logger.name)
+    log_q.exception('[QUEUE] Exception test: %s', log_q.logger.name)
+    log_q.stop_listener()
 
 
 if __name__ == "__main__":

test-run/logging_handling/log_queue_legacy.py

@@ -0,0 +1,31 @@
+"""
+Log logging_handling.log testing
+"""
+
+# import atexit
+from pathlib import Path
+from multiprocessing import Queue
+# this is for testing only
+from queue_logger.log_queue import QueueLogger
+
+
+def main():
+    """
+    Log testing
+    """
+    script_path: Path = Path(__file__).resolve().parent
+
+    log_queue: 'Queue[str]' = Queue()
+    log_q_legacy = QueueLogger(
+        log_file=script_path.joinpath('log', 'test_queue_legacy.log'),
+        log_name="Test Log Queue",
+        log_queue=log_queue
+    )
+    log_q_legacy.mlog.info('Log test: %s', 'Queue Legacy')
+    # log_q.stop_listener()
+
+
+if __name__ == "__main__":
+    main()
+
+# __END__

tests/unit/string_handling/test_timestamp_strings.py

@@ -82,7 +82,7 @@ class TestTimestampStrings:
 
     def test_class_default_timezone(self):
         """Test that class default timezone is correctly set"""
-        assert TimestampStrings.time_zone == 'Asia/Tokyo'
+        assert TimestampStrings.TIME_ZONE == 'Asia/Tokyo'
 
     def test_none_timezone_uses_default(self):
         """Test that passing None for timezone uses class default"""

uv.lock

@@ -44,7 +44,7 @@ wheels = [
 
 [[package]]
 name = "corelibs"
-version = "0.9.0"
+version = "0.10.0"
 source = { editable = "." }
 dependencies = [
     { name = "jmespath" },