feat(logger): implement addFilter/removeFilter to address static typing errors

aws_lambda_powertools/logging/logger.py

@@ -507,7 +507,7 @@ def exception(
         self,
         msg: object,
         *args,
-        exc_info=True,
+        exc_info: logging._ExcInfoType = True,
         stack_info: bool = False,
         stacklevel: int = 2,
         extra: Optional[Mapping[str, object]] = None,
@@ -683,6 +683,12 @@ def setLevel(self, level: Union[str, int, None]) -> None:
     def addHandler(self, handler: logging.Handler) -> None:
         return self._logger.addHandler(handler)
 
+    def addFilter(self, filter: logging._FilterType) -> None:  # noqa: A002 # filter built-in usage
+        return self._logger.addFilter(filter)
+
+    def removeFilter(self, filter: logging._FilterType) -> None:  # noqa: A002 # filter built-in usage
+        return self._logger.removeFilter(filter)
+
     @property
     def registered_handler(self) -> logging.Handler:
         """Convenience property to access the first logger handler"""

tests/functional/test_logger.py

@@ -5,6 +5,7 @@
 import logging
 import random
 import re
+import secrets
 import string
 import sys
 import warnings
@@ -1076,3 +1077,28 @@ def test_log_level_advanced_logging_controler_warning_different_log_levels_using
 
     # THEN Logger must be INFO because it takes precedence over POWERTOOLS_LOG_LEVEL
     assert logger.log_level == logging.INFO
+
+
+def test_logger_add_remove_filter(stdout, service_name):
+    # GIVEN a Logger with a custom logging filter
+    class ApiKeyFilter(logging.Filter):  # NOSONAR  # need filter to test actual impl.
+        def filter(self, record):
+            if getattr(record, "api_key", None):
+                record.api_key = "REDACTED"
+
+            return True
+
+    redact_api_key_filter = ApiKeyFilter()
+    logger = Logger(service=service_name, stream=stdout)
+    logger.addFilter(redact_api_key_filter)
+
+    # WHEN a new log statement is issued
+    # AND another log statement is issued after filter is removed
+    logger.info("filtered", api_key=secrets.token_urlsafe())
+    logger.removeFilter(redact_api_key_filter)
+    logger.info("unfiltered", api_key=secrets.token_urlsafe())
+
+    # THEN logging filter should be called and mutate the log record accordingly
+    log = capture_multiple_logging_statements_output(stdout)
+    assert log[0]["api_key"] == "REDACTED"
+    assert log[1]["api_key"] != "REDACTED"