PyPI/CoreLibs-PyPI-All
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Log level setter now uses LoggingLevel for levels, set/get log level

Browse Source 
Add flush for queue flushing

Add set/get level for handler

Allow adding handlers during launch, handlers cannot be added afterwards at the moment

Add testing for LoggingLevel enum
This commit is contained in:
Clemens Schwaighofer
2025-07-11 15:35:34 +09:00
parent 2a248bd249
commit 3f9f2ceaac
6 changed files with 545 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

223
src/corelibs/logging_handling/log.py
View File

@@ -7,6 +7,7 @@ attach "init_worker_logging" with the set log_queue
import re
import logging.handlers
import logging
import time
from pathlib import Path
from typing import Mapping, TextIO, TypedDict, Any, TYPE_CHECKING, cast
from corelibs.logging_handling.logging_level_handling.logging_level import LoggingLevel
@@ -16,6 +17,7 @@ if TYPE_CHECKING:
from multiprocessing import Queue
# MARK: Log settings TypedDict
class LogSettings(TypedDict):
"""
log settings
@@ -23,8 +25,8 @@ class LogSettings(TypedDict):
Arguments:
TypedDict {_type_} -- _description_
"""
log_level_console: str
log_level_file: str
log_level_console: LoggingLevel
log_level_file: LoggingLevel
console_enabled: bool
console_color_output_enabled: bool
add_start_info: bool
@@ -32,6 +34,7 @@ class LogSettings(TypedDict):
log_queue: 'Queue[str] | None'
# MARK: Custom color filter
class CustomConsoleFormatter(logging.Formatter):
"""
Custom formatter with colors for console output
@@ -68,6 +71,7 @@ class CustomConsoleFormatter(logging.Formatter):
return f"{color}{message}{reset}"
# MARK: Log class
class Log:
"""
logger setup
@@ -77,11 +81,13 @@ class Log:
SPACER_CHAR: str = '='
SPACER_LENGTH: int = 32
# default logging level
DEFAULT_LOG_LEVEL: str = 'WARNING'
DEFAULT_LOG_LEVEL: LoggingLevel = LoggingLevel.WARNING
DEFAULT_LOG_LEVEL_FILE: LoggingLevel = LoggingLevel.DEBUG
DEFAULT_LOG_LEVEL_CONSOLE: LoggingLevel = LoggingLevel.WARNING
# default settings
DEFAULT_LOG_SETTINGS: LogSettings = {
"log_level_console": "WARNING",
"log_level_file": "DEBUG",
"log_level_console": LoggingLevel.WARNING,
"log_level_file": LoggingLevel.DEBUG,
"console_enabled": True,
"console_color_output_enabled": True,
"add_start_info": True,
@@ -89,11 +95,13 @@ class Log:
"log_queue": None,
}
# MARK: constructor
def __init__(
self,
log_path: Path,
log_name: str,
log_settings: dict[str, 'str | bool | None | Queue[str]'] | LogSettings | None = None,
log_settings: dict[str, 'LoggingLevel | str | bool | None | Queue[str]'] | LogSettings | None = None,
other_handlers: dict[str, Any] | None = None
):
# add new level for EXCEPTION
logging.addLevelName(LoggingLevel.EXCEPTION.value, 'EXCEPTION')
@@ -122,13 +130,20 @@ class Log:
# setup handlers
# NOTE if console with color is set first, some of the color formatting is set
# in the file writer too, for the ones where color is set BEFORE the format
self.handlers: list[logging.StreamHandler[TextIO] | logging.handlers.TimedRotatingFileHandler] = [
# file handler, always
self.__file_handler(self.log_settings['log_level_file'], log_path)
]
# Any is logging.StreamHandler, logging.FileHandler and all logging.handlers.*
self.handlers: dict[str, Any] = {}
self.add_handler('file_handler', self.__create_time_rotating_file_handler(
self.log_settings['log_level_file'], log_path)
)
if self.log_settings['console_enabled']:
# console
self.handlers.append(self.__console_handler(self.log_settings['log_level_console']))
self.add_handler('stream_handler', self.__create_console_handler(
self.log_settings['log_level_console'])
)
# add other handlers,
if other_handlers is not None:
for handler_key, handler in other_handlers.items():
self.add_handler(handler_key, handler)
# init listener if we have a log_queue set
self.__init_listener(self.log_settings['log_queue'])
@@ -138,6 +153,7 @@ class Log:
if self.log_settings['add_start_info'] is True:
self.break_line('START')
# MARK: deconstructor
def __del__(self):
"""
Call when class is destroyed, make sure the listender is closed or else we throw a thread error
@@ -146,9 +162,10 @@ class Log:
self.break_line('END')
self.stop_listener()
# MARK: parse log settings
def __parse_log_settings(
self,
log_settings: dict[str, 'str | bool | None | Queue[str]'] | LogSettings | None
log_settings: dict[str, 'LoggingLevel | str | bool | None | Queue[str]'] | LogSettings | None
) -> LogSettings:
# skip with defaul it not set
if log_settings is None:
@@ -160,11 +177,11 @@ class Log:
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 = self.DEFAULT_LOG_SETTINGS.get(
if not self.validate_log_level(__log_level := log_settings.get(__log_entry, '')):
__log_level = self.DEFAULT_LOG_SETTINGS.get(
__log_entry, self.DEFAULT_LOG_LEVEL
)
default_log_settings[__log_entry] = str(_log_level)
default_log_settings[__log_entry] = LoggingLevel.from_any(__log_level)
# check bool
for __log_entry in [
"console_enabled",
@@ -187,10 +204,32 @@ class Log:
def __filter_exceptions(self, record: logging.LogRecord) -> bool:
return record.levelname != "EXCEPTION"
def __console_handler(self, log_level_console: str = 'WARNING') -> logging.StreamHandler[TextIO]:
# MARK: add a handler
def add_handler(
self,
handler_name: str,
handler: Any
) -> bool:
"""
Add a log handler to the handlers dict
Arguments:
handler_name {str} -- _description_
handler {Any} -- _description_
"""
if self.handlers.get(handler_name):
return False
# TODO: handler must be some handler type, how to check?
self.handlers[handler_name] = handler
return True
# MARK: console handler
def __create_console_handler(
self, log_level_console: LoggingLevel = LoggingLevel.WARNING, filter_exceptions: bool = True
) -> logging.StreamHandler[TextIO]:
# console logger
if not isinstance(getattr(logging, log_level_console.upper(), None), int):
log_level_console = 'WARNING'
if not self.validate_log_level(log_level_console):
log_level_console = self.DEFAULT_LOG_LEVEL_CONSOLE
console_handler = logging.StreamHandler()
# format layouts
format_string = (
@@ -206,21 +245,31 @@ class Log:
formatter_console = CustomConsoleFormatter(format_string, datefmt=format_date)
else:
formatter_console = logging.Formatter(format_string, datefmt=format_date)
console_handler.setLevel(log_level_console)
console_handler.setLevel(log_level_console.name)
console_handler.set_name('console')
# do not show exceptions logs on console
console_handler.addFilter(self.__filter_exceptions)
if filter_exceptions:
console_handler.addFilter(self.__filter_exceptions)
console_handler.setFormatter(formatter_console)
return console_handler
def __file_handler(self, log_level_file: str, log_path: Path) -> logging.handlers.TimedRotatingFileHandler:
# MARK: file handler
def __create_time_rotating_file_handler(
self, log_level_file: LoggingLevel, log_path: Path,
when: str = "D", interval: int = 1, backup_count: int = 0
) -> logging.handlers.TimedRotatingFileHandler:
# file logger
if not isinstance(getattr(logging, log_level_file.upper(), None), int):
log_level_file = 'DEBUG'
# when: S/M/H/D/W0-W6/midnight
# interval: how many, 1D = every day
# backup_count: how many old to keep, 0 = all
if not self.validate_log_level(log_level_file):
log_level_file = self.DEFAULT_LOG_LEVEL_FILE
file_handler = logging.handlers.TimedRotatingFileHandler(
filename=log_path,
encoding="utf-8",
when="D",
interval=1
when=when,
interval=interval,
backupCount=backup_count
)
formatter_file_handler = logging.Formatter(
(
@@ -239,10 +288,12 @@ class Log:
),
datefmt="%Y-%m-%dT%H:%M:%S",
)
file_handler.setLevel(log_level_file)
file_handler.set_name('file_timed_rotate')
file_handler.setLevel(log_level_file.name)
file_handler.setFormatter(formatter_file_handler)
return file_handler
# MARK: init listener
def __init_listener(self, log_queue: 'Queue[str] | None' = None):
"""
If we have a Queue option start the logging queue
@@ -255,11 +306,12 @@ class Log:
self.log_queue = log_queue
self.listener = logging.handlers.QueueListener(
self.log_queue,
*self.handlers,
*self.handlers.values(),
respect_handler_level=True
)
self.listener.start()
# MARK: init main log
def __init_log(self, log_name: str) -> None:
"""
Initialize the main loggger
@@ -271,7 +323,7 @@ class Log:
self.logger = logging.getLogger(log_name)
# add all the handlers
if queue_handler is None:
for handler in self.handlers:
for handler in self.handlers.values():
self.logger.addHandler(handler)
else:
self.logger.addHandler(queue_handler)
@@ -279,6 +331,7 @@ class Log:
# log level filtering is done per handler
self.logger.setLevel(logging.DEBUG)
# MARK: init logger for Fork/Thread
@staticmethod
def init_worker_logging(log_queue: 'Queue[str]') -> logging.Logger:
"""
@@ -287,6 +340,7 @@ class Log:
queue_handler = logging.handlers.QueueHandler(log_queue)
# getLogger call MUST be WITHOUT and logger name
root_logger = logging.getLogger()
# base logging level, filtering is done in the handlers
root_logger.setLevel(logging.DEBUG)
root_logger.handlers.clear()
root_logger.addHandler(queue_handler)
@@ -296,22 +350,7 @@ class Log:
return root_logger
def stop_listener(self):
"""
stop the listener
"""
if self.listener is not None:
self.listener.stop()
def break_line(self, info: str = "BREAK"):
"""
add a break line as info level
Keyword Arguments:
info {str} -- _description_ (default: {"BREAK"})
"""
self.logger.info("[%s] %s>", info, self.SPACER_CHAR * self.SPACER_LENGTH)
# MARK: log message
def exception(self, msg: object, *args: object, extra: Mapping[str, object] | None = None) -> None:
"""
log on exceotion level
@@ -323,6 +362,100 @@ class Log:
"""
self.logger.log(LoggingLevel.EXCEPTION.value, msg, *args, exc_info=True, extra=extra)
# MARK: break line
def break_line(self, info: str = "BREAK"):
"""
add a break line as info level
Keyword Arguments:
info {str} -- _description_ (default: {"BREAK"})
"""
self.logger.info("[%s] %s>", info, self.SPACER_CHAR * self.SPACER_LENGTH)
# MARK: queue handling
def flush(self, handler_name: str | None = None, timeout: float = 2.0) -> bool:
"""
Flush all pending messages
Keyword Arguments:
handler_name {str | None} -- _description_ (default: {None})
timeout {float} -- _description_ (default: {2.0})
Returns:
bool -- _description_
"""
if not self.listener or not self.log_queue:
return False
try:
# Wait for queue to be processed
start_time = time.time()
while not self.log_queue.empty() and (time.time() - start_time) < timeout:
time.sleep(0.01)
# Flush all handlers or handler given
if handler_name:
try:
self.handlers[handler_name].flush()
except IndexError:
pass
else:
for handler in self.handlers.values():
handler.flush()
except OSError:
return False
return True
def stop_listener(self):
"""
stop the listener
"""
if self.listener is not None:
self.flush()
self.listener.stop()
# MARK: log level handling
def set_log_level(self, handler_name: str, log_level: LoggingLevel) -> bool:
"""
set the logging level
Arguments:
handler {str} -- _description_
log_level {LoggingLevel} -- _description_
Returns:
bool -- _description_
"""
try:
# flush queue befoe changing logging level
self.flush(handler_name)
self.handlers[handler_name].setLevel(log_level.name)
return True
except IndexError:
self.logger.error('Handler %s not found, cannot change log level', handler_name)
return False
except AttributeError:
self.logger.error(
'Cannot change to log level %s for handler %s, log level invalid',
LoggingLevel.name, handler_name
)
return False
def get_log_level(self, handler_name: str) -> LoggingLevel:
"""
gett the logging level for a handler
Arguments:
handler_name {str} -- _description_
Returns:
LoggingLevel -- _description_
"""
try:
return self.handlers[handler_name]
except IndexError:
return LoggingLevel.NOTSET
@staticmethod
def validate_log_level(log_level: Any) -> bool:
"""
@@ -355,6 +488,6 @@ class Log:
try:
return LoggingLevel.from_any(log_level).value
except ValueError:
return LoggingLevel.from_string(Log.DEFAULT_LOG_LEVEL).value
return LoggingLevel.from_string(Log.DEFAULT_LOG_LEVEL.name).value
# __END__

15
src/corelibs/logging_handling/logging_level_handling/logging_level.py
View File

@@ -30,6 +30,8 @@ class LoggingLevel(Enum):
return cls[level_str.upper()]
except KeyError as e:
raise ValueError(f"Invalid log level: {level_str}") from e
except AttributeError as e:
raise ValueError(f"Invalid log level: {level_str}") from e
@classmethod
def from_int(cls, level_int: int):
@@ -41,7 +43,18 @@ class LoggingLevel(Enum):
@classmethod
def from_any(cls, level_any: Any):
"""Convert either str or int"""
"""
Convert any vale
if self LoggingLevel return as is, else try to convert from int or string
Arguments:
level_any {Any} -- _description_
Returns:
_type_ -- _description_
"""
if isinstance(level_any, LoggingLevel):
return level_any
if isinstance(level_any, int):
return cls.from_int(level_any)
return cls.from_string(level_any)