Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 6 additions & 14 deletions src/zenlib/logging/loggermixin.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,9 @@
__author__ = "desultory"
__version__ = "1.3.1"
__version__ = "1.4.0"

from logging import Logger, getLogger

from zenlib.logging.utils import add_handler_if_not_exists, handle_additional_logging, log_init
from zenlib.logging.utils import add_handler_if_not_exists, log_init


class LoggerMixIn:
Expand All @@ -15,30 +15,22 @@ class LoggerMixIn:
Otherwise, the log level is not set, and the logger will use the parent's level.
"""

def init_logger(self, args, kwargs):
def init_logger(self, args, kwargs) -> None:
# Get the parent logger from the root if one was not passed
parent_logger = kwargs.pop("logger") if isinstance(kwargs.get("logger"), Logger) else getLogger()

# Get a child logger from the parent logger, set self.logger
self.logger = parent_logger.getChild(self.__class__.__name__)

if log_level := kwargs.pop("_log_level", None):
# Set the logger's level if _log_level is passed
self.logger.setLevel(log_level)
elif log_bump := kwargs.pop("_log_bump", None):
# get the parent logger's level, or the parent's parent logger's level
parent_logger = self.logger.parent
while parent_logger:
if parent_logger.level != 0:
break
parent_logger = parent_logger.parent

self.logger.setLevel(parent_logger.level + log_bump)
# bump the class's logger using the level of the parent logger
self.logger.setLevel(parent_logger.getEffectiveLevel() + log_bump)

# Add a colored stream handler if one does not exist
add_handler_if_not_exists(self.logger)

# Log class init if _log_init is passed
log_init(self, args, kwargs)

# Add logging to _log_setattr if set
handle_additional_logging(self, kwargs)
49 changes: 15 additions & 34 deletions src/zenlib/logging/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
from zenlib.logging.colorlognameformatter import ColorLognameFormatter


def _logger_has_handler(logger):
def _logger_has_handler(logger: Logger | None) -> bool:
"""Checks if a logger or its parents has a handler already"""
while logger:
if logger.handlers:
Expand All @@ -14,9 +14,9 @@ def _logger_has_handler(logger):
return False


def add_handler_if_not_exists(logger):
def add_handler_if_not_exists(logger: Logger) -> None:
"""Adds a ColorLognameFormatter handler to the logger if it doesn't have a handler already
Coloring is diabled by the _ZENLIB_COLOR_TEXT variable in the colorize function
Coloring is disabled by the _ZENLIB_COLOR_TEXT variable in the colorize function
"""
if _logger_has_handler(logger):
return
Expand All @@ -25,52 +25,33 @@ def add_handler_if_not_exists(logger):
stream_handler.setFormatter(formatter)

logger.addHandler(stream_handler)
logger.info("Added default handler to logger: %s", logger)
logger.info(f"Added default handler to logger: {logger}")


def log_init(self, args, kwargs):
def log_init(self, args, kwargs) -> None:
"""If _log_init is in the kwargs and set to True, logs init args, kwargs, class name, and version"""
class_name = self.__class__.__name__
logger = self.logger
if not kwargs.pop("_log_init", False):
return logger.log(5, "Init logging disabled for class: %s", class_name)
return logger.log(5, f"Init logging disabled for class: {class_name}")

logger.info("Initializing class: %s", class_name)
logger.info(f"Initializing class: {class_name}")

if args:
logger.debug("[%s] Init args: %s" % (class_name, args))
logger.debug(f"[{class_name}] Init args: {args}")
if kwargs:
logger.debug("[%s] Init kwargs: %s" % (class_name, kwargs))
logger.debug(f"[{class_name}] Init kwargs: {kwargs}")

package_name = self.__module__.split(".")[0]
try:
logger.info("[%s] Package version: %s" % (package_name, version(package_name)))
logger.info(f"[{package_name}] Package version: {version(package_name)}")
except (NameError, PackageNotFoundError) as ex:
if ex.msg != "No package metadata was found for builtins":
logger.debug("[%s] Package version not found for: %s" % (class_name, package_name))
if str(ex) == "No package metadata was found for builtins":
package_name = "builtins"
logger.debug(f"[{class_name}] Package version not found for: {package_name}")

if module_version := getattr(modules.get(self.__module__), "__version__", None):
logger.info("[%s] Module version: %s" % (self.__module__, module_version))
logger.info(f"[{self.__module__}] Module version: {module_version}")

if class_version := getattr(self, "__version__", None):
logger.info("[%s] Class version: %s" % (class_name, class_version))


def handle_additional_logging(self, kwargs):
"""Sets __setattr__ to log_setattr if _log_setattr is in the kwargs and set to True"""
if kwargs.pop("_log_setattr", False):
setattr(self, "__setattr__", log_setattr)


def log_setattr(self, name, value):
"""Logs when an attribute is set"""
super().__setattr__(name, value)
# check if the logger is defined
if not isinstance(self.logger, Logger):
raise ValueError("Logger is not defined")

# Log containers or strings with newlines on a new line
if isinstance(value, list) or isinstance(value, dict) or isinstance(value, str) and "\n" in value:
self.logger.log(5, "Setattr '%s' to:\n%s" % (name, getattr(self, name)))
else:
self.logger.log(5, "Setattr '%s' to: %s" % (name, getattr(self, name)))
logger.info(f"[{class_name}] Class version: {class_version}")
19 changes: 10 additions & 9 deletions src/zenlib/util/handle_plural.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,45 +2,46 @@
__version__ = "2.2.1"

from collections.abc import KeysView, ValuesView
from typing import Any, Callable


def handle_plural(function, log_level=10):
def handle_plural(function: Callable[..., Any], log_level: int = 5) -> Callable[..., Any]:
"""
Wraps functions to take a list/dict and iterate over it.
The last passed argument should be the iterable.
Logs using the logger attribute if it exists.
"""

def wrapper(self, *args, **kwargs):
def log(msg, level=log_level):
def wrapper(self: Any, *args: Any, **kwargs: Any) -> Any:
def log(msg: str, level: int = log_level):
if hasattr(self, "logger"):
self.logger.log(level, msg)

if len(args) == 1:
focus_arg = args[0]
other_args = tuple()
other_args: tuple[Any, ...] = tuple()
else:
focus_arg = args[-1]
other_args = args[:-1]

if isinstance(focus_arg, list) and not isinstance(focus_arg, str):
log("Expanding list: %s" % focus_arg)
log(f"Expanding list: {focus_arg}")
for item in focus_arg:
function(self, *(other_args + (item,)), **kwargs)
elif isinstance(focus_arg, set):
log("Expanding set: %s" % focus_arg)
log(f"Expanding set: {focus_arg}")
for item in focus_arg:
function(self, *(other_args + (item,)), **kwargs)
elif isinstance(focus_arg, ValuesView):
log("Expanding dict values: %s" % focus_arg)
log(f"Expanding dict values: {focus_arg}")
for value in focus_arg:
function(self, *(other_args + (value,)), **kwargs)
elif isinstance(focus_arg, KeysView):
log("Expanding dict keys: %s" % focus_arg)
log(f"Expanding dict keys: {focus_arg}")
for key in focus_arg:
function(self, *(other_args + (key,)), **kwargs)
elif isinstance(focus_arg, dict):
log("Expanding dict: %s" % focus_arg)
log(f"Expanding dict: {focus_arg}")
for key, value in focus_arg.items():
function(
self,
Expand Down
Loading