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

aws_lambda_powertools/logging/utils.py

@@ -0,0 +1,81 @@
+import logging
+from typing import Callable, List, Optional, Set, Union
+
+from .logger import Logger
+
+
+def copy_config_to_registered_loggers(
+    source_logger: Logger,
+    log_level: Optional[str] = None,
+    exclude: Optional[Set[str]] = None,
+    include: Optional[Set[str]] = None,
+) -> None:
+
+    """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
+    """
+
+    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
+    else:
+        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, level)
+
+
+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: 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: 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: Logger, logger: logging.Logger, level: Union[int, str]) -> None:
+    logger.handlers = []
+    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}")

docs/core/logger.md

@@ -1062,6 +1062,24 @@ for the given name and level to the logging module. By default, this logs all bo
         return response.get("Buckets", [])
     ```
 
+**How can I enable powertools logging for imported libraries?**
+
+You can copy the Logger setup to all or sub-sets of registered external loggers. Use the `copy_config_to_registered_logger` method to do this. By default all registered loggers will be modified. You can change this behaviour by providing `include` and `exclude` attributes. You can also provide optional `log_level` attribute external loggers will be configured with.
+
+
+```python hl_lines="10" title="Cloning Logger config to all other registered standard loggers"
+import logging
+
+from aws_lambda_powertools import Logger
+from aws_lambda_powertools.logging import utils
+
+logger = Logger()
+
+external_logger = logging.logger()
+
+utils.copy_config_to_registered_loggers(source_logger=logger)
+external_logger.info("test message")
+
 **What's the difference between `append_keys` and `extra`?**
 
 Keys added with `append_keys` will persist across multiple log messages while keys added via `extra` will only be available in a given log message operation.

tests/functional/test_logger_utils.py

@@ -0,0 +1,193 @@
+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
+        WARNING = 30
+        CRITICAL = 50
+
+    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
+
+
+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