Add custom log level atrribute

Author: Michal Ploski

aws_lambda_powertools/logging/utils.py

@@ -1,65 +1,81 @@
 import logging
-from typing import Callable, List, Optional, TypeVar
+from typing import Callable, List, Optional, Set, Union
 
 from .logger import Logger
 
-PowertoolsLogger = TypeVar("PowertoolsLogger", bound=Logger)
-
 
 def copy_config_to_registered_loggers(
-    source_logger: PowertoolsLogger,
-    exclude: Optional[List[str]] = None,
-    include: Optional[List[str]] = None,
+    source_logger: Logger,
+    log_level: Optional[str] = None,
+    exclude: Optional[Set[str]] = None,
+    include: Optional[Set[str]] = None,
 ) -> None:
-    """Enable powertools logging for imported libraries.
 
-    Attach source logger handlers to external loggers.
-    Modify logger level based on source logger attribute.
-    Ensure powertools logger itself is excluded from registered list.
+    """Copies source Logger level and handler to all registered loggers for consistent formatting.
+
+    Parameters
+    ----------
+    source_logger : Logger
+        Powertools Logger to copy configuration from
+    log_level : str, optional
+        Logging level to set to registered loggers, by default uses source_logger logging level
+    include : Optional[Set[str]], optional
+        List of logger names to include, by default all registered loggers are included
+    exclude : Optional[Set[str]], optional
+        List of logger names to exclude, by default None
     """
 
-    if include and not exclude:
-        loggers = include
-        filter_func = _include_registered_loggers_filter
-    elif include and exclude:
-        exclude = [source_logger.name, *exclude]
-        loggers = list(set(include) - set(exclude))
+    level = log_level or source_logger.level
+
+    # Assumptions: Only take parent loggers not children (dot notation rule)
+    # Steps:
+    # 1. Default operation: Include all registered loggers
+    # 2. Only include set? Only add Loggers in the list and ignore all else
+    # 3. Include and exclude set? Add Logger if it’s in include and not in exclude
+    # 4. Only exclude set? Ignore Logger in the excluding list
+
+    # Exclude source logger by default
+    if exclude:
+        exclude.add(source_logger.name)
+    else:
+        exclude = set(source_logger.name)
+
+    # Prepare loggers set
+    if include:
+        loggers = include.difference(exclude)
         filter_func = _include_registered_loggers_filter
-    elif not include and exclude:
-        loggers = [source_logger.name, *exclude]
-        filter_func = _exclude_registered_loggers_filter
     else:
-        loggers = [source_logger.name]
+        loggers = exclude
         filter_func = _exclude_registered_loggers_filter
 
     registered_loggers = _find_registered_loggers(source_logger, loggers, filter_func)
     for logger in registered_loggers:
-        _configure_logger(source_logger, logger)
+        _configure_logger(source_logger, logger, level)
 
 
-def _include_registered_loggers_filter(loggers: List[str]):
+def _include_registered_loggers_filter(loggers: Set[str]):
     return [logging.getLogger(name) for name in logging.root.manager.loggerDict if "." not in name and name in loggers]
 
 
-def _exclude_registered_loggers_filter(loggers: List[str]) -> List[logging.Logger]:
+def _exclude_registered_loggers_filter(loggers: Set[str]) -> List[logging.Logger]:
     return [
         logging.getLogger(name) for name in logging.root.manager.loggerDict if "." not in name and name not in loggers
     ]
 
 
 def _find_registered_loggers(
-    source_logger: PowertoolsLogger, loggers: List[str], filter_func: Callable
+    source_logger: Logger, loggers: Set[str], filter_func: Callable[[Set[str]], List[logging.Logger]]
 ) -> List[logging.Logger]:
     """Filter root loggers based on provided parameters."""
     root_loggers = filter_func(loggers)
     source_logger.debug(f"Filtered root loggers: {root_loggers}")
     return root_loggers
 
 
-def _configure_logger(source_logger: PowertoolsLogger, logger: logging.Logger) -> None:
+def _configure_logger(source_logger: Logger, logger: logging.Logger, level: Union[int, str]) -> None:
     logger.handlers = []
-    logger.setLevel(source_logger.level)
-    source_logger.debug(f"Logger {logger} reconfigured to use logging level {source_logger.level}")
+    logger.setLevel(level)
+    source_logger.debug(f"Logger {logger} reconfigured to use logging level {level}")
     for source_handler in source_logger.handlers:
         logger.addHandler(source_handler)
         source_logger.debug(f"Logger {logger} reconfigured to use {source_handler}")

tests/functional/test_logger_utils.py

@@ -21,6 +21,8 @@ def log_level():
     class LogLevel(Enum):
         NOTSET = 0
         INFO = 20
+        WARNING = 30
+        CRITICAL = 50
 
     return LogLevel
 
@@ -83,7 +85,7 @@ def test_copy_config_to_ext_loggers_include(stdout, logger, log_level):
     powertools_logger = Logger(service=service_name(), level=log_level.INFO.value, stream=stdout)
 
     # WHEN configuration copied from powertools logger to ALL external loggers AND our external logger used
-    utils.copy_config_to_registered_loggers(source_logger=powertools_logger, include=[logger.name])
+    utils.copy_config_to_registered_loggers(source_logger=powertools_logger, include={logger.name})
     logger.info(msg)
     log = capture_logging_output(stdout)
 
@@ -103,7 +105,7 @@ def test_copy_config_to_ext_loggers_wrong_include(stdout, logger, log_level):
     powertools_logger = Logger(service=service_name(), level=log_level.INFO.value, stream=stdout)
 
     # WHEN configuration copied from powertools logger to ALL external loggers AND our external logger used
-    utils.copy_config_to_registered_loggers(source_logger=powertools_logger, include=["non-existing-logger"])
+    utils.copy_config_to_registered_loggers(source_logger=powertools_logger, include={"non-existing-logger"})
 
     # THEN
     assert not logger.handlers
@@ -116,7 +118,7 @@ def test_copy_config_to_ext_loggers_exclude(stdout, logger, log_level):
     powertools_logger = Logger(service=service_name(), level=log_level.INFO.value, stream=stdout)
 
     # WHEN configuration copied from powertools logger to ALL external loggers AND our external logger used
-    utils.copy_config_to_registered_loggers(source_logger=powertools_logger, exclude=[logger.name])
+    utils.copy_config_to_registered_loggers(source_logger=powertools_logger, exclude={logger.name})
 
     # THEN
     assert not logger.handlers
@@ -134,7 +136,7 @@ def test_copy_config_to_ext_loggers_include_exclude(stdout, logger, log_level):
 
     # WHEN configuration copied from powertools logger to ALL external loggers AND our external logger used
     utils.copy_config_to_registered_loggers(
-        source_logger=powertools_logger, include=[logger_1.name, logger_2.name], exclude=[logger_1.name]
+        source_logger=powertools_logger, include={logger_1.name, logger_2.name}, exclude={logger_1.name}
     )
     logger_2.info(msg)
     log = capture_logging_output(stdout)
@@ -164,3 +166,28 @@ def test_copy_config_to_ext_loggers_clean_old_handlers(stdout, logger, log_level
     assert len(logger.handlers) == 1
     assert type(logger.handlers[0]) is logging.StreamHandler
     assert type(logger.handlers[0].formatter) is formatter.LambdaPowertoolsFormatter
+
+
+def test_copy_config_to_ext_loggers_custom_log_level(stdout, logger, log_level):
+
+    msg = "test message"
+
+    # GIVEN a external logger and powertools logger initialized
+    logger = logger()
+    powertools_logger = Logger(service=service_name(), level=log_level.CRITICAL.value, stream=stdout)
+    level = log_level.WARNING.name
+
+    # WHEN configuration copied from powertools logger to ALL external loggers
+    # AND our external logger used with custom log_level
+    utils.copy_config_to_registered_loggers(source_logger=powertools_logger, include={logger.name}, log_level=level)
+    logger.warning(msg)
+    log = capture_logging_output(stdout)
+
+    # THEN
+    assert len(logger.handlers) == 1
+    assert type(logger.handlers[0]) is logging.StreamHandler
+    assert type(logger.handlers[0].formatter) is formatter.LambdaPowertoolsFormatter
+    assert powertools_logger.level == log_level.CRITICAL.value
+    assert logger.level == log_level.WARNING.value
+    assert log["message"] == msg
+    assert log["level"] == log_level.WARNING.name