feat(logger): clone powertools logger config to any Python logger

aws_lambda_powertools/logging/utils.py

@@ -0,0 +1,65 @@
+import logging
+from typing import Callable, List, Optional, TypeVar
+
+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,
+) -> 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.
+    """
+
+    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))
+        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]
+        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)
+
+
+def _include_registered_loggers_filter(loggers: List[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]:
+    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
+) -> 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:
+    logger.handlers = []
+    logger.setLevel(source_logger.level)
+    source_logger.debug(f"Logger {logger} reconfigured to use logging level {source_logger.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

@@ -0,0 +1,166 @@
+import io
+import json
+import logging
+import random
+import string
+from enum import Enum
+
+import pytest
+
+from aws_lambda_powertools import Logger
+from aws_lambda_powertools.logging import formatter, utils
+
+
+@pytest.fixture
+def stdout():
+    return io.StringIO()
+
+
+@pytest.fixture
+def log_level():
+    class LogLevel(Enum):
+        NOTSET = 0
+        INFO = 20
+
+    return LogLevel
+
+
+@pytest.fixture
+def logger(stdout, log_level):
+    def _logger():
+        logging.basicConfig(stream=stdout, level=log_level.NOTSET.value)
+        logger = logging.getLogger(name=service_name())
+        return logger
+
+    return _logger
+
+
+def capture_logging_output(stdout):
+    return json.loads(stdout.getvalue().strip())
+
+
+def capture_multiple_logging_statements_output(stdout):
+    return [json.loads(line.strip()) for line in stdout.getvalue().split("\n") if line]
+
+
+def service_name():
+    chars = string.ascii_letters + string.digits
+    return "".join(random.SystemRandom().choice(chars) for _ in range(15))
+
+
+def test_copy_config_to_ext_loggers(stdout, logger, log_level):
+
+    msg = "test message"
+
+    # GIVEN a external logger and powertools logger initialized
+    logger_1 = logger()
+    logger_2 = logger()
+
+    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)
+    logger_1.info(msg)
+    logger_2.info(msg)
+    logs = capture_multiple_logging_statements_output(stdout)
+
+    # THEN
+    for index, logger in enumerate([logger_1, logger_2]):
+        assert len(logger.handlers) == 1
+        assert type(logger.handlers[0]) is logging.StreamHandler
+        assert type(logger.handlers[0].formatter) is formatter.LambdaPowertoolsFormatter
+        assert logger.level == log_level.INFO.value
+        assert logs[index]["message"] == msg
+        assert logs[index]["level"] == log_level.INFO.name
+
+
+def test_copy_config_to_ext_loggers_include(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.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])
+    logger.info(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 logger.level == log_level.INFO.value
+    assert log["message"] == msg
+    assert log["level"] == log_level.INFO.name
+
+
+def test_copy_config_to_ext_loggers_wrong_include(stdout, logger, log_level):
+
+    # GIVEN a external logger and powertools logger initialized
+    logger = logger()
+    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"])
+
+    # THEN
+    assert not logger.handlers
+
+
+def test_copy_config_to_ext_loggers_exclude(stdout, logger, log_level):
+
+    # GIVEN a external logger and powertools logger initialized
+    logger = logger()
+    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])
+
+    # THEN
+    assert not logger.handlers
+
+
+def test_copy_config_to_ext_loggers_include_exclude(stdout, logger, log_level):
+
+    msg = "test message"
+
+    # GIVEN a external logger and powertools logger initialized
+    logger_1 = logger()
+    logger_2 = logger()
+
+    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_1.name, logger_2.name], exclude=[logger_1.name]
+    )
+    logger_2.info(msg)
+    log = capture_logging_output(stdout)
+
+    # THEN
+    assert not logger_1.handlers
+    assert len(logger_2.handlers) == 1
+    assert type(logger_2.handlers[0]) is logging.StreamHandler
+    assert type(logger_2.handlers[0].formatter) is formatter.LambdaPowertoolsFormatter
+    assert logger_2.level == log_level.INFO.value
+    assert log["message"] == msg
+    assert log["level"] == log_level.INFO.name
+
+
+def test_copy_config_to_ext_loggers_clean_old_handlers(stdout, logger, log_level):
+
+    # GIVEN a external logger with handler and powertools logger initialized
+    logger = logger()
+    handler = logging.FileHandler("logfile")
+    logger.addHandler(handler)
+    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)
+
+    # THEN
+    assert len(logger.handlers) == 1
+    assert type(logger.handlers[0]) is logging.StreamHandler
+    assert type(logger.handlers[0].formatter) is formatter.LambdaPowertoolsFormatter