619 lines
19 KiB
Python
619 lines
19 KiB
Python
from __future__ import annotations
|
|
|
|
from functools import wraps
|
|
from logging import *
|
|
from logging import __all__ as logging__all__
|
|
from logging import handlers
|
|
from pathlib import Path as Path_
|
|
import os
|
|
import time
|
|
|
|
__all__ = tuple(logging__all__ + [
|
|
'Decorator',
|
|
'FormatterColor',
|
|
'StreamHandlerFormatted',
|
|
'add_default_handler',
|
|
'critical',
|
|
'debug',
|
|
'error',
|
|
'get_default_logger',
|
|
'get_default_logger_child',
|
|
'get_formatted_logger',
|
|
'info',
|
|
'log',
|
|
'set_default_level',
|
|
'set_default_logger',
|
|
'warning',
|
|
])
|
|
__version__ = '0.1.3'
|
|
__author__ = 'Eishausener <code@eishausener.de>'
|
|
__name__ = 'eh-logger'
|
|
|
|
##########
|
|
# config #
|
|
##########
|
|
|
|
|
|
# debug print (not formatted)
|
|
_DEBUG = False
|
|
|
|
|
|
class DEFAULT:
|
|
# -- Default Logger Name -- #
|
|
LOGGER_NAME = 'eh_logging'
|
|
# -- Logger Space -- #
|
|
SPACE_LOGGER_NAME = 11
|
|
SPACE_LEVEL = 8
|
|
SPACE_TIME = 19
|
|
|
|
|
|
####################
|
|
# helper functions #
|
|
####################
|
|
|
|
|
|
def Path(*args, **kwargs):
|
|
"""
|
|
Returns the Path from the pathlib module as String
|
|
:param args:
|
|
:param kwargs:
|
|
:return:
|
|
"""
|
|
return str(Path_(*args, **kwargs))
|
|
|
|
|
|
def print_debug(*args, **kwargs):
|
|
"""
|
|
only print to console if _DEBUG is True
|
|
:param args:
|
|
:param kwargs:
|
|
"""
|
|
if _DEBUG:
|
|
print(*args, **kwargs)
|
|
|
|
|
|
class _DefaultLogger:
|
|
"""
|
|
Helper class to hold only one instance of the default logger
|
|
"""
|
|
logger: Logger | None = None
|
|
|
|
def get(self):
|
|
print_debug(f'[get] addr: {id(self.logger)}, logger: {self.logger}')
|
|
return self.logger
|
|
|
|
def set(self, logger: Logger) -> None:
|
|
print_debug(f'[set] addr: {id(self.logger)}, logger: {self.logger}')
|
|
self.logger = logger
|
|
|
|
|
|
_default_logger = _DefaultLogger()
|
|
|
|
|
|
def _init():
|
|
"""
|
|
Initialize a logger with StreamHandler and set as default logger (without formatting).
|
|
If during the initialization process of the module console outputs must be done, this logger is used. In the end,
|
|
it is replaced with a formatted logger
|
|
"""
|
|
_changed = False
|
|
_logger = _default_logger.get()
|
|
if not _logger:
|
|
print_debug('[DEBUG] creating new logger')
|
|
_logger = getLogger(DEFAULT.LOGGER_NAME)
|
|
_changed = True
|
|
|
|
if not _logger.handlers:
|
|
print_debug(f'[DEBUG] adding stream handler to logger. (addr: {id(_logger)})')
|
|
_handler = StreamHandler()
|
|
_handler.name = 'eh_logging-stream-helper' # should not be displayed. only for internal use
|
|
_logger.addHandler(_handler)
|
|
_changed = True
|
|
|
|
if _changed:
|
|
_default_logger.set(_logger)
|
|
|
|
|
|
_init()
|
|
|
|
|
|
############################
|
|
# default logger functions #
|
|
############################
|
|
|
|
|
|
def get_default_logger() -> Logger:
|
|
"""
|
|
Returns the default logger
|
|
:return:
|
|
"""
|
|
return _default_logger.get()
|
|
|
|
|
|
def set_default_logger(logger: Logger) -> None:
|
|
"""
|
|
Sets the default logger
|
|
:param logger:
|
|
:return:
|
|
"""
|
|
_default_logger.set(logger)
|
|
|
|
|
|
def add_default_handler(handler: Handler) -> None:
|
|
"""
|
|
Adds the handler to the default logger
|
|
:param handler:
|
|
:return:
|
|
"""
|
|
logger = _default_logger.get()
|
|
logger.addHandler(handler)
|
|
_default_logger.set(logger)
|
|
|
|
|
|
def set_default_level(level: int) -> None:
|
|
"""
|
|
Sets the default logger level
|
|
:param level:
|
|
:return:
|
|
"""
|
|
logger = _default_logger.get()
|
|
logger.setLevel(level)
|
|
_default_logger.set(logger)
|
|
|
|
|
|
def get_default_logger_child(suffix: str) -> Logger:
|
|
"""
|
|
Returns a child logger of the default logger
|
|
:param suffix:
|
|
:return:
|
|
"""
|
|
logger = _default_logger.get()
|
|
return logger.getChild(suffix=suffix)
|
|
|
|
|
|
#########################
|
|
# root logger functions #
|
|
#########################
|
|
|
|
|
|
def log(level, msg, *args, **kwargs):
|
|
"""
|
|
Log 'msg % args' with the integer severity 'level' on the default logger.
|
|
"""
|
|
_logger = _default_logger.get()
|
|
_logger.log(level, msg, *args, **kwargs)
|
|
|
|
|
|
def debug(msg, *args, **kwargs):
|
|
"""
|
|
Log a message with severity 'DEBUG' on the default logger.
|
|
"""
|
|
_logger = _default_logger.get()
|
|
_logger.debug(msg, *args, **kwargs)
|
|
|
|
|
|
def info(msg, *args, **kwargs):
|
|
"""
|
|
Log a message with severity 'INFO' on the default logger.
|
|
"""
|
|
_logger = _default_logger.get()
|
|
_logger.info(msg, *args, **kwargs)
|
|
|
|
|
|
def warning(msg, *args, **kwargs):
|
|
"""
|
|
Log a message with severity 'WARNING' on the default logger.
|
|
"""
|
|
_logger = _default_logger.get()
|
|
_logger.warning(msg, *args, **kwargs)
|
|
|
|
|
|
def error(msg, *args, **kwargs):
|
|
"""
|
|
Log a message with severity 'ERROR' on the default logger.
|
|
"""
|
|
_logger = _default_logger.get()
|
|
_logger.error(msg, *args, **kwargs)
|
|
|
|
|
|
def critical(msg, *args, **kwargs):
|
|
"""
|
|
Log a message with severity 'CRITICAL' on the default logger.
|
|
"""
|
|
_logger = _default_logger.get()
|
|
_logger.critical(msg, *args, **kwargs)
|
|
|
|
|
|
#############
|
|
# Decorator #
|
|
#############
|
|
|
|
|
|
def _decorator_log_by_level(logger, arg, kwarg, return_value, return_value_type, decimal_places,
|
|
level, function, *args, **kwargs):
|
|
"""
|
|
Executes the function and returns the result of the function and does all the logging stuff
|
|
:param logger:
|
|
:param arg:
|
|
:param kwarg:
|
|
:param return_value:
|
|
:param return_value_type:
|
|
:param decimal_places:
|
|
:param level:
|
|
:param function:
|
|
:param args:
|
|
:param kwargs:
|
|
:return:
|
|
"""
|
|
if logger is None:
|
|
logger = _default_logger.get()
|
|
fname = function.__name__
|
|
msg = f'Function \x1b[3m{fname}\x1b[0m started'
|
|
if arg:
|
|
msg += f' with args: {args}'
|
|
if kwarg and not arg:
|
|
msg += f' with kwargs: {kwargs}'
|
|
if kwarg and arg:
|
|
msg += f' and with kwargs: {kwargs}'
|
|
logger.log(level, msg)
|
|
time_start = time.time()
|
|
value = function(*args, **kwargs)
|
|
time_needed = time.time() - time_start
|
|
if isinstance(time_needed, int):
|
|
time_needed = round(time_needed, decimal_places)
|
|
msg = f'Function \x1b[3m{fname}\x1b[0m finished in {time_needed}sec'
|
|
if return_value_type:
|
|
msg += f' with return value type {type(value)}'
|
|
if return_value and return_value_type:
|
|
msg += f' and value {value}'
|
|
if return_value and not return_value_type:
|
|
msg += f' with return value {value}'
|
|
logger.log(level, msg)
|
|
return value
|
|
|
|
|
|
class Decorator:
|
|
"""
|
|
Decorators to log
|
|
"""
|
|
|
|
@staticmethod
|
|
def func_logger(logger: Logger = None, arg: bool = False, kwarg: bool = False, return_value: bool = False,
|
|
return_value_type: bool = False, decimal_places: int = 2, level: int = DEBUG):
|
|
"""
|
|
Decorator to add logging to a function
|
|
:param logger:
|
|
:param arg:
|
|
:param kwarg:
|
|
:param return_value:
|
|
:param return_value_type:
|
|
:param decimal_places:
|
|
:param level:
|
|
:return:
|
|
"""
|
|
|
|
def decorator(function):
|
|
@wraps(function)
|
|
def wrapper(*args, **kwargs):
|
|
nonlocal logger
|
|
return _decorator_log_by_level(logger, arg, kwarg, return_value, return_value_type, decimal_places,
|
|
level, function, *args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
return decorator
|
|
|
|
@staticmethod
|
|
def debug(logger: Logger = None, arg: bool = False, kwarg: bool = False, return_value: bool = False,
|
|
return_value_type: bool = False, decimal_places: int = 2):
|
|
"""
|
|
Decorator to add logging to a function with the Level DEBUG
|
|
:param logger:
|
|
:param arg:
|
|
:param kwarg:
|
|
:param return_value:
|
|
:param return_value_type:
|
|
:param decimal_places:
|
|
:return:
|
|
"""
|
|
|
|
def decorator(function):
|
|
@wraps(function)
|
|
def wrapper(*args, **kwargs):
|
|
nonlocal logger
|
|
return _decorator_log_by_level(logger, arg, kwarg, return_value, return_value_type, decimal_places,
|
|
DEBUG, function, *args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
return decorator
|
|
|
|
@staticmethod
|
|
def info(logger=None, arg: bool = False, kwarg: bool = False, return_value: bool = False,
|
|
return_value_type: bool = False, decimal_places: int = 2):
|
|
"""
|
|
Decorator to add logging to a function with the Level INFO
|
|
:param logger:
|
|
:param arg:
|
|
:param kwarg:
|
|
:param return_value:
|
|
:param return_value_type:
|
|
:param decimal_places:
|
|
:return:
|
|
"""
|
|
|
|
def decorator(function):
|
|
@wraps(function)
|
|
def wrapper(*args, **kwargs):
|
|
nonlocal logger
|
|
return _decorator_log_by_level(logger, arg, kwarg, return_value, return_value_type, decimal_places,
|
|
INFO, function, *args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
return decorator
|
|
|
|
@staticmethod
|
|
def warning(logger=None, arg: bool = False, kwarg: bool = False, return_value: bool = False,
|
|
return_value_type: bool = False, decimal_places: int = 2):
|
|
"""
|
|
Decorator to add logging to a function with the Level WARNING
|
|
:param logger:
|
|
:param arg:
|
|
:param kwarg:
|
|
:param return_value:
|
|
:param return_value_type:
|
|
:param decimal_places:
|
|
:return:
|
|
"""
|
|
|
|
def decorator(function):
|
|
@wraps(function)
|
|
def wrapper(*args, **kwargs):
|
|
nonlocal logger
|
|
return _decorator_log_by_level(logger, arg, kwarg, return_value, return_value_type, decimal_places,
|
|
WARNING, function, *args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
return decorator
|
|
|
|
@staticmethod
|
|
def error(logger=None, arg: bool = False, kwarg: bool = False, return_value: bool = False,
|
|
return_value_type: bool = False, decimal_places: int = 2):
|
|
"""
|
|
Decorator to add logging to a function with the Level ERROR
|
|
:param logger:
|
|
:param arg:
|
|
:param kwarg:
|
|
:param return_value:
|
|
:param return_value_type:
|
|
:param decimal_places:
|
|
:return:
|
|
"""
|
|
|
|
def decorator(function):
|
|
@wraps(function)
|
|
def wrapper(*args, **kwargs):
|
|
nonlocal logger
|
|
return _decorator_log_by_level(logger, arg, kwarg, return_value, return_value_type, decimal_places,
|
|
ERROR, function, *args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
return decorator
|
|
|
|
@staticmethod
|
|
def critical(logger=None, arg: bool = False, kwarg: bool = False, return_value: bool = False,
|
|
return_value_type: bool = False, decimal_places: int = 2):
|
|
"""
|
|
Decorator to add logging to a function with the Level CRITICAL
|
|
:param logger:
|
|
:param arg:
|
|
:param kwarg:
|
|
:param return_value:
|
|
:param return_value_type:
|
|
:param decimal_places:
|
|
:return:
|
|
"""
|
|
|
|
def decorator(function):
|
|
@wraps(function)
|
|
def wrapper(*args, **kwargs):
|
|
nonlocal logger
|
|
return _decorator_log_by_level(logger, arg, kwarg, return_value, return_value_type, decimal_places,
|
|
CRITICAL, function, *args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
return decorator
|
|
|
|
|
|
#############
|
|
# Formatter #
|
|
#############
|
|
|
|
|
|
class FormatterColor(Formatter):
|
|
space_logger_name: int | None
|
|
space_level: int | None
|
|
space_time: int | None
|
|
time_format: str
|
|
|
|
def __init__(self, space_logger_name: int = None, space_level: int = None, space_time: int = None,
|
|
time_format: str = '%Y-%m-%d %H:%M:%S'):
|
|
"""
|
|
|
|
:param space_logger_name:
|
|
:param space_level:
|
|
:param space_time:
|
|
:param time_format:
|
|
"""
|
|
super().__init__()
|
|
|
|
temp_logger = getLogger('FormatterColor')
|
|
# space_logger_name
|
|
if space_logger_name is not None:
|
|
try:
|
|
self.space_logger_name = int(space_logger_name)
|
|
except ValueError:
|
|
self.space_logger_name = DEFAULT.SPACE_LOGGER_NAME
|
|
temp_logger.warning(
|
|
f'the type of space_logger_name must be int not {type(space_logger_name)}! Set to default '
|
|
f'value of 10')
|
|
else:
|
|
self.space_logger_name = DEFAULT.SPACE_LOGGER_NAME
|
|
# space_level
|
|
if space_level is not None:
|
|
try:
|
|
self.space_level = int(space_level)
|
|
except ValueError:
|
|
self.space_level = DEFAULT.SPACE_LEVEL
|
|
temp_logger.warning(
|
|
f'the type of space_level must be int not {type(space_logger_name)}! Set to default value of 8')
|
|
else:
|
|
self.space_level = DEFAULT.SPACE_LEVEL
|
|
# space_time
|
|
if space_time is not None:
|
|
try:
|
|
self.space_time = int(space_time)
|
|
except ValueError:
|
|
self.space_time = DEFAULT.SPACE_TIME
|
|
temp_logger.warning(
|
|
f'the type of space_time must be int not {type(space_logger_name)}! Set to default value of 19')
|
|
else:
|
|
self.space_time = DEFAULT.SPACE_TIME
|
|
# time_format
|
|
if time_format is not None:
|
|
self.time_format = time_format
|
|
|
|
# https://en.wikipedia.org/wiki/ANSI_escape_code
|
|
self.color_by_level = (
|
|
(DEBUG, '\033[38;5;85;1m'), # 85 Bold
|
|
(INFO, '\033[96;1m'), # LIGHTCYAN_EX Bold
|
|
(WARNING, '\033[93;1m'), # LIGHTYELLOW_EX Bold
|
|
(ERROR, '\033[31;1m'), # RED Bold
|
|
(CRITICAL, '\033[31;1;3;5m'), # RED Bold/Italic/blinking
|
|
)
|
|
|
|
self.formats = {
|
|
level: Formatter(
|
|
f'\x1b[30;1m%(asctime)-{self.space_time}s\x1b[0m {colour}%(levelname)-{self.space_level}s\x1b[0m '
|
|
f'\x1b[36m%(name)-{self.space_logger_name}s\x1b[0m %(message)s',
|
|
self.time_format,
|
|
)
|
|
for level, colour in self.color_by_level
|
|
}
|
|
|
|
def format(self, record: LogRecord):
|
|
formatter = self.formats.get(record.levelno)
|
|
if formatter is None:
|
|
formatter = self.formats[DEBUG]
|
|
output = formatter.format(record)
|
|
return output
|
|
|
|
|
|
class StreamHandlerFormatted(StreamHandler):
|
|
|
|
def __init__(self, level: int = None, stream=None):
|
|
"""
|
|
when the logger this handler will be added,
|
|
has a higher level, then this handler,
|
|
it will not log (this is normal)
|
|
the best is to set the logger to 1 (not 0)
|
|
|
|
:param level:
|
|
:param stream:
|
|
"""
|
|
super().__init__(stream=stream)
|
|
# set Formatter
|
|
self.formatter = FormatterColor()
|
|
|
|
# set level if it is available
|
|
if level is not None:
|
|
self.level = level
|
|
|
|
|
|
def get_formatted_logger(name: str, console=True, console_level: int = INFO, file=False, file_path: str = None,
|
|
file_level: int = WARNING, file_backup_count: int = None,
|
|
file_rotate: str = 'midnight', file_interval: int = 1, logger_level: int = 1,
|
|
space_logger_name: int = None, space_level: int = None, space_time: int = None,
|
|
time_format: str = '%Y-%m-%d %H:%M:%S') -> Logger:
|
|
"""
|
|
|
|
|
|
:param name: the name of the logger
|
|
:param console: if a StreamHandler should be added
|
|
:param console_level: the minimum level who will be in the console
|
|
:param file: if a FileHandler should be added
|
|
:param file_path: the full path of the logfile
|
|
:param file_level: the minimum level who will be in the logfile
|
|
:param file_backup_count:
|
|
:param file_rotate:
|
|
:param file_interval:
|
|
:param logger_level: should be lower or equal than handlers of it
|
|
:param space_logger_name:
|
|
:param space_level:
|
|
:param space_time:
|
|
:param time_format:
|
|
:return:
|
|
"""
|
|
logger = getLogger(name)
|
|
logger.setLevel(logger_level)
|
|
if console:
|
|
handler = StreamHandler()
|
|
handler.setFormatter(FormatterColor(space_logger_name, space_level, space_time, time_format))
|
|
handler.setLevel(console_level)
|
|
logger.addHandler(handler)
|
|
if console_level < logger_level:
|
|
get_default_logger_child('get_formatted_logger').warning(
|
|
f'the console (level {console_level}) will not log all, because of the logger level is {logger_level}')
|
|
if file:
|
|
if file_path is None:
|
|
get_default_logger_child('get_formatted_logger').warning(f'file_path is None. No file handler added')
|
|
else:
|
|
# check if log dir exists if not create it
|
|
file_path = os.path.abspath(file_path)
|
|
file_name = os.path.split(file_path)[-1]
|
|
if not file_name.endswith('.log'):
|
|
file_name += '.log'
|
|
file_path += '.log'
|
|
get_default_logger_child('get_formatted_logger').warning(f'file_path should be end with .log. adding '
|
|
f'.log to {file_name}')
|
|
if not os.path.exists((path := os.path.dirname(file_path))):
|
|
os.makedirs(path)
|
|
if file_backup_count is not None:
|
|
handler = handlers.TimedRotatingFileHandler(file_path, backupCount=file_backup_count,
|
|
when=file_rotate, interval=file_interval)
|
|
handler.namer = lambda name_logfile: name_logfile.replace('.log', '') + '.log'
|
|
else:
|
|
handler = FileHandler(file_path)
|
|
handler.setFormatter(FormatterColor())
|
|
handler.setLevel(file_level)
|
|
logger.addHandler(handler)
|
|
if file_level < logger_level:
|
|
get_default_logger_child('get_formatted_logger').warning(
|
|
f'the file (level {file_level}) will not log all, because of the logger level is {logger_level}')
|
|
|
|
return logger
|
|
|
|
|
|
def _init_default_logger():
|
|
"""
|
|
Initialize the default logger with a formatted logger
|
|
:return:
|
|
"""
|
|
_logger = getLogger(DEFAULT.LOGGER_NAME)
|
|
_formatter = FormatterColor()
|
|
_handler = StreamHandler()
|
|
_handler.setFormatter(_formatter)
|
|
_handler.name = 'eh_logging-stream-default' # should not be displayed. only for internal use
|
|
# remove all handlers to prevent duplicate handlers
|
|
for handler in _logger.handlers:
|
|
_logger.removeHandler(handler)
|
|
_logger.addHandler(_handler)
|
|
_default_logger.set(_logger)
|
|
|
|
|
|
_init_default_logger()
|