refactor(logger): remove subclassing and move unnecessary APIs (#2334)

Co-authored-by: Leandro Damascena <leandro.damascena@gmail.com>

Author: heitorlessa

aws_lambda_powertools/logging/compat.py

@@ -0,0 +1,51 @@
+"""Maintenance: We can drop this upon Py3.7 EOL. It's a backport for "location" key to work."""
+from __future__ import annotations
+
+import io
+import logging
+import os
+import traceback
+
+
+def findCaller(stack_info=False, stacklevel=2):  # pragma: no cover
+    """
+    Find the stack frame of the caller so that we can note the source
+    file name, line number and function name.
+    """
+    f = logging.currentframe()  # noqa: VNE001
+    # On some versions of IronPython, currentframe() returns None if
+    # IronPython isn't run with -X:Frames.
+    if f is None:
+        return "(unknown file)", 0, "(unknown function)", None
+    while stacklevel > 0:
+        next_f = f.f_back
+        if next_f is None:
+            ## We've got options here.
+            ## If we want to use the last (deepest) frame:
+            break
+            ## If we want to mimic the warnings module:
+            # return ("sys", 1, "(unknown function)", None) # noqa: E800
+            ## If we want to be pedantic:  # noqa: E800
+            # raise ValueError("call stack is not deep enough") # noqa: E800
+        f = next_f  # noqa: VNE001
+        if not _is_internal_frame(f):
+            stacklevel -= 1
+    co = f.f_code
+    sinfo = None
+    if stack_info:
+        with io.StringIO() as sio:
+            sio.write("Stack (most recent call last):\n")
+            traceback.print_stack(f, file=sio)
+            sinfo = sio.getvalue()
+            if sinfo[-1] == "\n":
+                sinfo = sinfo[:-1]
+    return co.co_filename, f.f_lineno, co.co_name, sinfo
+
+
+# The following is based on warnings._is_internal_frame. It makes sure that
+# frames of the import mechanism are skipped when logging at module level and
+# using a stacklevel value greater than one.
+def _is_internal_frame(frame):  # pragma: no cover
+    """Signal whether the frame is a CPython or logging module internal."""
+    filename = os.path.normcase(frame.f_code.co_filename)
+    return filename == logging._srcfile or ("importlib" in filename and "_bootstrap" in filename)

aws_lambda_powertools/logging/logger.py

@@ -2,12 +2,10 @@
 
 import functools
 import inspect
-import io
 import logging
 import os
 import random
 import sys
-import traceback
 from typing import (
     IO,
     TYPE_CHECKING,
@@ -25,6 +23,8 @@
 
 import jmespath
 
+from aws_lambda_powertools.logging import compat
+
 from ..shared import constants
 from ..shared.functions import (
     extract_event_from_common_models,
@@ -66,12 +66,7 @@ def _is_cold_start() -> bool:
     return cold_start
 
 
-# PyCharm does not support autocomplete via getattr
-# so we need to return to subclassing removed in #97
-# All methods/properties continue to be proxied to inner logger
-# https://github.com/awslabs/aws-lambda-powertools-python/issues/107
-# noinspection PyRedeclaration
-class Logger(logging.Logger):  # lgtm [py/missing-call-to-init]
+class Logger:
     """Creates and setups a logger to format statements in JSON.
 
     Includes service name and any additional key=value into logs
@@ -238,7 +233,6 @@ def __init__(
         self.logger_handler = logger_handler or logging.StreamHandler(stream)
         self.log_uncaught_exceptions = log_uncaught_exceptions
 
-        self.log_level = self._get_log_level(level)
         self._is_deduplication_disabled = resolve_truthy_env_var_choice(
             env=os.getenv(constants.LOGGER_LOG_DEDUPLICATION_ENV, "false")
         )
@@ -258,7 +252,7 @@ def __init__(
             "use_rfc3339": use_rfc3339,
         }
 
-        self._init_logger(formatter_options=formatter_options, **kwargs)
+        self._init_logger(formatter_options=formatter_options, log_level=level, **kwargs)
 
         if self.log_uncaught_exceptions:
             logger.debug("Replacing exception hook")
@@ -277,11 +271,11 @@ def _get_logger(self):
         """Returns a Logger named {self.service}, or {self.service.filename} for child loggers"""
         logger_name = self.service
         if self.child:
-            logger_name = f"{self.service}.{self._get_caller_filename()}"
+            logger_name = f"{self.service}.{_get_caller_filename()}"
 
         return logging.getLogger(logger_name)
 
-    def _init_logger(self, formatter_options: Optional[Dict] = None, **kwargs):
+    def _init_logger(self, formatter_options: Optional[Dict] = None, log_level: Union[str, int, None] = None, **kwargs):
         """Configures new logger"""
 
         # Skip configuration if it's a child logger or a pre-configured logger
@@ -293,13 +287,13 @@ def _init_logger(self, formatter_options: Optional[Dict] = None, **kwargs):
         if self.child or is_logger_preconfigured:
             return
 
+        self._logger.setLevel(self._determine_log_level(log_level))
         self._configure_sampling()
-        self._logger.setLevel(self.log_level)
         self._logger.addHandler(self.logger_handler)
         self.structure_logs(formatter_options=formatter_options, **kwargs)
 
         # Maintenance: We can drop this upon Py3.7 EOL. It's a backport for "location" key to work
-        self._logger.findCaller = self.findCaller
+        self._logger.findCaller = compat.findCaller
 
         # Pytest Live Log feature duplicates log records for colored output
         # but we explicitly add a filter for log deduplication.
@@ -329,7 +323,7 @@ def _configure_sampling(self):
         try:
             if self.sampling_rate and random.random() <= float(self.sampling_rate):
                 logger.debug("Setting log level to Debug due to sampling rate")
-                self.setLevel(logging.DEBUG)
+                self._logger.setLevel(logging.DEBUG)
         except ValueError:
             raise InvalidLoggerSamplingRateError(
                 f"Expected a float value ranging 0 to 1, but received {self.sampling_rate} instead."
@@ -445,19 +439,6 @@ def decorate(event, context, *args, **kwargs):
 
         return decorate
 
-    def setLevel(self, level: Union[str, int]):
-        """
-        Set the logging level for the logger.
-
-        Parameters:
-        -----------
-        level str | int
-            The level to set. Can be a string representing the level name: 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'
-            or an integer representing the level value: 10 for 'DEBUG', 20 for 'INFO', 30 for 'WARNING', 40 for 'ERROR', 50 for 'CRITICAL'. # noqa: E501
-        """
-        self.log_level = level
-        self._logger.setLevel(level)
-
     def info(
         self,
         msg: object,
@@ -584,17 +565,6 @@ def append_keys(self, **additional_keys):
     def remove_keys(self, keys: Iterable[str]):
         self.registered_formatter.remove_keys(keys)
 
-    @property
-    def registered_handler(self) -> logging.Handler:
-        """Convenience property to access logger handler"""
-        handlers = self._logger.parent.handlers if self.child else self._logger.handlers
-        return handlers[0]
-
-    @property
-    def registered_formatter(self) -> BasePowertoolsFormatter:
-        """Convenience property to access logger formatter"""
-        return self.registered_handler.formatter  # type: ignore
-
     def structure_logs(self, append: bool = False, formatter_options: Optional[Dict] = None, **keys):
         """Sets logging formatting to JSON.
 
@@ -663,8 +633,38 @@ def get_correlation_id(self) -> Optional[str]:
             return self.registered_formatter.log_format.get("correlation_id")
         return None
 
+    @property
+    def registered_handler(self) -> logging.Handler:
+        """Convenience property to access the first logger handler"""
+        handlers = self._logger.parent.handlers if self.child else self._logger.handlers
+        return handlers[0]
+
+    @property
+    def registered_formatter(self) -> BasePowertoolsFormatter:
+        """Convenience property to access the first logger formatter"""
+        return self.registered_handler.formatter  # type: ignore[return-value]
+
+    @property
+    def log_level(self) -> int:
+        return self._logger.level
+
+    @property
+    def name(self) -> str:
+        return self._logger.name
+
+    @property
+    def handlers(self) -> List[logging.Handler]:
+        """List of registered logging handlers
+
+        Notes
+        -----
+
+        Looking for the first configured handler? Use registered_handler property instead.
+        """
+        return self._logger.handlers
+
     @staticmethod
-    def _get_log_level(level: Union[str, int, None]) -> Union[str, int]:
+    def _determine_log_level(level: Union[str, int, None]) -> Union[str, int]:
         """Returns preferred log level set by the customer in upper case"""
         if isinstance(level, int):
             return level
@@ -675,51 +675,6 @@ def _get_log_level(level: Union[str, int, None]) -> Union[str, int]:
 
         return log_level.upper()
 
-    @staticmethod
-    def _get_caller_filename():
-        """Return caller filename by finding the caller frame"""
-        # Current frame         => _get_logger()
-        # Previous frame        => logger.py
-        # Before previous frame => Caller
-        frame = inspect.currentframe()
-        caller_frame = frame.f_back.f_back.f_back
-        return caller_frame.f_globals["__name__"]
-
-    # Maintenance: We can drop this upon Py3.7 EOL. It's a backport for "location" key to work
-    def findCaller(self, stack_info=False, stacklevel=2):  # pragma: no cover
-        """
-        Find the stack frame of the caller so that we can note the source
-        file name, line number and function name.
-        """
-        f = logging.currentframe()  # noqa: VNE001
-        # On some versions of IronPython, currentframe() returns None if
-        # IronPython isn't run with -X:Frames.
-        if f is None:
-            return "(unknown file)", 0, "(unknown function)", None
-        while stacklevel > 0:
-            next_f = f.f_back
-            if next_f is None:
-                ## We've got options here.
-                ## If we want to use the last (deepest) frame:
-                break
-                ## If we want to mimic the warnings module:
-                # return ("sys", 1, "(unknown function)", None) # noqa: E800
-                ## If we want to be pedantic:  # noqa: E800
-                # raise ValueError("call stack is not deep enough") # noqa: E800
-            f = next_f  # noqa: VNE001
-            if not _is_internal_frame(f):
-                stacklevel -= 1
-        co = f.f_code
-        sinfo = None
-        if stack_info:
-            with io.StringIO() as sio:
-                sio.write("Stack (most recent call last):\n")
-                traceback.print_stack(f, file=sio)
-                sinfo = sio.getvalue()
-                if sinfo[-1] == "\n":
-                    sinfo = sinfo[:-1]
-        return co.co_filename, f.f_lineno, co.co_name, sinfo
-
 
 def set_package_logger(
     level: Union[str, int] = logging.DEBUG,
@@ -760,16 +715,16 @@ def set_package_logger(
     logger.addHandler(handler)
 
 
-# Maintenance: We can drop this upon Py3.7 EOL. It's a backport for "location" key to work
-# The following is based on warnings._is_internal_frame. It makes sure that
-# frames of the import mechanism are skipped when logging at module level and
-# using a stacklevel value greater than one.
-def _is_internal_frame(frame):  # pragma: no cover
-    """Signal whether the frame is a CPython or logging module internal."""
-    filename = os.path.normcase(frame.f_code.co_filename)
-    return filename == logging._srcfile or ("importlib" in filename and "_bootstrap" in filename)
-
-
 def log_uncaught_exception_hook(exc_type, exc_value, exc_traceback, logger: Logger):
     """Callback function for sys.excepthook to use Logger to log uncaught exceptions"""
     logger.exception(exc_value, exc_info=(exc_type, exc_value, exc_traceback))  # pragma: no cover
+
+
+def _get_caller_filename():
+    """Return caller filename by finding the caller frame"""
+    # Current frame         => _get_logger()
+    # Previous frame        => logger.py
+    # Before previous frame => Caller
+    frame = inspect.currentframe()
+    caller_frame = frame.f_back.f_back.f_back
+    return caller_frame.f_globals["__name__"]

aws_lambda_powertools/logging/utils.py

@@ -25,7 +25,7 @@ def copy_config_to_registered_loggers(
     exclude : Optional[Set[str]], optional
         List of logger names to exclude, by default None
     """
-    level = log_level or source_logger.level
+    level = log_level or source_logger.log_level
 
     # Assumptions: Only take parent loggers not children (dot notation rule)
     # Steps: