Convert runtime annotations to plugin and additional testing for actions plugin

Author: crimsonknave

annotated_logger/__init__.py

@@ -24,7 +24,7 @@
 from makefun import wraps
 
 from annotated_logger.filter import AnnotatedFilter
-from annotated_logger.plugins import BasePlugin
+from annotated_logger.plugins import BasePlugin, RuntimeAnnotationsPlugin
 
 if TYPE_CHECKING:  # pragma: no cover
     from collections.abc import MutableMapping
@@ -255,8 +255,8 @@ class AnnotatedLogger:
     Args:
     ----
         annotations: Dictionary of annotations to be added to every log message
-        runtime_annotations: dictionary of method references to be called when
-            a log message is emitted
+        runtime_annotations: [Deprecated] Dictionary of method references to be called
+            when a log message is emitted. Use the `RuntimeAnnotationsPlugin` instead.
         plugins: list of plugins to use
 
     Methods:
@@ -313,9 +313,11 @@ def __init__(  # noqa: PLR0913
         self.logger_base = logging.getLogger(self.logger_root_name)
         self.logger_base.setLevel(self.log_level)
         self.annotations = annotations or {}
-        self.runtime_annotations = runtime_annotations or {}
         self.plugins = [BasePlugin()]
         self.plugins.extend(plugins)
+        # Preserve the `runtime_annotations` param for backwards compat
+        if runtime_annotations:
+            self.plugins.append(RuntimeAnnotationsPlugin(runtime_annotations))
         if formatter and config:
             msg = "Cannot pass both formatter and config."
             raise ValueError(msg)
@@ -385,7 +387,6 @@ def generate_filter(
         return AnnotatedFilter(
             annotations=annotations,
             class_annotations=class_annotations,
-            runtime_annotations=self.runtime_annotations,
             plugins=self.plugins,
         )
 

annotated_logger/filter.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import logging
+from copy import copy
 from typing import Any
 
 import annotated_logger
@@ -14,29 +15,23 @@ class AnnotatedFilter(logging.Filter):
     def __init__(
         self,
         annotations: Annotations | None = None,
-        runtime_annotations: Annotations | None = None,
         class_annotations: Annotations | None = None,
         plugins: list[annotated_logger.BasePlugin] | None = None,
     ) -> None:
         """Store the annotations, attributes and plugins."""
         self.annotations = annotations or {}
         self.class_annotations = class_annotations or {}
-        self.runtime_annotations = runtime_annotations or {}
         self.plugins = plugins or [annotated_logger.BasePlugin()]
 
         # This allows plugins to determine what fields were added by the user
         # vs the ones native to the log record
         # TODO(crimsonknave): Make a test for this # noqa: TD003, FIX002
         self.base_attributes = logging.makeLogRecord({}).__dict__  # pragma: no mutate
 
-    def _all_annotations(self, record: logging.LogRecord) -> Annotations:
+    def _all_annotations(self) -> Annotations:
         annotations = {}
-        # Using copy might be better, but, we don't want to add
-        # the runtime annotations to the stored annotations
-        annotations.update(self.class_annotations)
-        annotations.update(self.annotations)
-        for key, function in self.runtime_annotations.items():
-            annotations[key] = function(record)
+        annotations.update(copy(self.class_annotations))
+        annotations.update(copy(self.annotations))
         annotations["annotated"] = True
         return annotations
 
@@ -48,7 +43,7 @@ def filter(self, record: logging.LogRecord) -> bool:
         sees it. Returning False from the filter method will stop the evaluation and
         the log record won't be emitted.
         """
-        record.__dict__.update(self._all_annotations(record))
+        record.__dict__.update(self._all_annotations())
         for plugin in self.plugins:
             try:
                 result = plugin.filter(record)

annotated_logger/plugins.py

@@ -2,7 +2,7 @@
 
 import contextlib
 import logging
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any, Callable
 
 from requests.exceptions import HTTPError
 
@@ -23,11 +23,29 @@ def uncaught_exception(
         self, exception: Exception, logger: AnnotatedAdapter
     ) -> AnnotatedAdapter:
         """Handle an uncaught excaption."""
-        logger.annotate(success=False)
-        logger.annotate(exception_title=str(exception))
+        if "success" not in logger.filter.annotations:
+            logger.annotate(success=False)
+        if "exception_title" not in logger.filter.annotations:
+            logger.annotate(exception_title=str(exception))
         return logger
 
 
+class RuntimeAnnotationsPlugin(BasePlugin):
+    """Plugin that sets annotations dynamically."""
+
+    def __init__(
+        self, runtime_annotations: dict[str, Callable[[logging.LogRecord], Any]]
+    ) -> None:
+        """Store the runtime annotations."""
+        self.runtime_annotations = runtime_annotations
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        """Add any configured runtime annotations."""
+        for key, function in self.runtime_annotations.items():
+            record.__dict__[key] = function(record)
+        return True
+
+
 class RequestsPlugin(BasePlugin):
     """Plugin for the requests library."""
 
@@ -182,7 +200,7 @@ def logging_config(self) -> dict[str, dict[str, object]]:
                     "level": "DEBUG",
                     "handlers": [
                         # This is from the default logging config
-                        "annotated_handler",
+                        #  "annotated_handler",
                         "actions_handler",
                     ],
                 },

example/actions.py

@@ -41,3 +41,8 @@ class ActionsExample:
     def first_step(self, annotated_logger: AnnotatedAdapter) -> None:
         """First step of your action."""
         annotated_logger.info("Step 1 running!")
+
+    @annotate_logs(_typing_requested=True)
+    def second_step(self, annotated_logger: AnnotatedAdapter) -> None:
+        """Second step of your action."""
+        annotated_logger.debug("Step 2 running!")

example/api.py

@@ -7,7 +7,7 @@
 from requests.models import Response
 
 from annotated_logger import AnnotatedAdapter, AnnotatedLogger
-from annotated_logger.plugins import RequestsPlugin
+from annotated_logger.plugins import RequestsPlugin, RuntimeAnnotationsPlugin
 
 
 def runtime(_record: logging.LogRecord) -> str:
@@ -17,8 +17,10 @@ def runtime(_record: logging.LogRecord) -> str:
 
 annotated_logger = AnnotatedLogger(
     annotations={"extra": "new data"},
-    runtime_annotations={"runtime": runtime},
-    plugins=[RequestsPlugin()],
+    plugins=[
+        RequestsPlugin(),
+        RuntimeAnnotationsPlugin({"runtime": runtime}),
+    ],
     log_level=logging.DEBUG,
     name="annotated_logger.api",
 )

example/calculator.py

@@ -7,6 +7,7 @@
     NameAdjusterPlugin,
     NestedRemoverPlugin,
     RemoverPlugin,
+    RuntimeAnnotationsPlugin,
 )
 
 
@@ -24,13 +25,13 @@ def runtime(_record: logging.LogRecord) -> str:
         "extra": "new data",
         "nested_extra": {"nested_key": {"double_nested_key": "value"}},
     },
-    runtime_annotations={"runtime": runtime},
     log_level=logging.DEBUG,
     plugins=[
         NameAdjusterPlugin(names=["joke"], prefix="cheezy_"),
         NameAdjusterPlugin(names=["power"], postfix="_overwhelming"),
         RemoverPlugin("taskName"),
         NestedRemoverPlugin(["double_nested_key"]),
+        RuntimeAnnotationsPlugin({"runtime": runtime}),
     ],
     name="annotated_logger.calculator",
 )

example/logging_config.py

@@ -2,7 +2,7 @@
 import logging.config
 
 from annotated_logger import AnnotatedAdapter, AnnotatedFilter, AnnotatedLogger
-from annotated_logger.plugins import BasePlugin, RenamerPlugin
+from annotated_logger.plugins import BasePlugin, RenamerPlugin, RuntimeAnnotationsPlugin
 
 # This logging config creates 4 loggers:
 # * A logger for "annotated_logger.logging_config", which logs all messages as json and
@@ -48,8 +48,10 @@
             "()": AnnotatedFilter,
             "annotations": {"decorated": False, "class_based_filter": True},
             "class_annotations": {},
-            "runtime_annotations": {"custom_runtime": lambda _record: True},
-            "plugins": [BasePlugin()],
+            "plugins": [
+                BasePlugin(),
+                RuntimeAnnotationsPlugin({"custom_runtime": lambda _record: True}),
+            ],
         },
     },
     "handlers": {
@@ -153,6 +155,10 @@ def runtime(_record: logging.LogRecord) -> str:
 
 annotated_logger = AnnotatedLogger(
     annotations={"hostname": "my-host"},
+    # This is deprecated, use the RuntimeAnnotationsPlugin instead.
+    # This param is kept for backwards compatibility and creates a
+    # RuntimeAnnotationsPlugin instead.
+    # This is left as an example and to provide test coverage.
     runtime_annotations={"runtime": runtime},
     plugins=[RenamerPlugin(time="created", lvl="levelname")],
     log_level=logging.DEBUG,

test/test_decorator.py

@@ -11,6 +11,7 @@
 import example.default
 import test.demo
 from annotated_logger import AnnotatedLogger
+from annotated_logger.plugins import RuntimeAnnotationsPlugin
 
 if TYPE_CHECKING:
     from annotated_logger.mocks import AnnotatedLogMock
@@ -217,10 +218,14 @@ def test_debug(self, annotated_logger_mock):
     def test_runtime_not_cached(self, annotated_logger_mock, mocker):
         runtime_mock = mocker.Mock(name="runtime_not_cached")
         runtime_mock.side_effect = ["first", "second", "third", "fourth"]
-        runtime_annotations = example.calculator.annotated_logger.runtime_annotations
-        example.calculator.annotated_logger.runtime_annotations = {
-            "runtime": runtime_mock
-        }
+        plugin = next(
+            plugin
+            for plugin in example.calculator.annotated_logger.plugins
+            if isinstance(plugin, RuntimeAnnotationsPlugin)
+        )
+
+        runtime_annotations = plugin.runtime_annotations
+        plugin.runtime_annotations = {"runtime": runtime_mock}
         # Need to use full path as that's what's reloaded. The other tests don't need
         # to as they're not mocking the runtime annotations
         calc = example.calculator.Calculator(12, 13)
@@ -243,7 +248,7 @@ def test_runtime_not_cached(self, annotated_logger_mock, mocker):
                 "runtime": "second",
             },
         )
-        example.calculator.annotated_logger.runtime_annotations = runtime_annotations
+        plugin.runtime_annotations = runtime_annotations
 
     def test_raises_type_error_with_too_few_args(self):
         calc = example.calculator.Calculator(12, 13)

test/test_plugins.py

@@ -236,12 +236,21 @@ def test_exclude_nested_fields(self, annotated_logger_mock):
         )
 
 
+@pytest.mark.usefixtures("_reload_actions")
 class TestGitHubActionsPlugin:
     def test_logs_normally(self, annotated_logger_mock):
         action = ActionsExample()
         action.first_step()
 
         annotated_logger_mock.assert_logged("info", "Step 1 running!")
 
-    def test_logs_actions_annotations(self, annotated_logger_mock, caplog):
-        pass
+    @pytest.mark.parametrize(
+        "annotated_logger_object", [logging.getLogger("annotated_logger.actions")]
+    )
+    def test_logs_actions_annotations(self, annotated_logger_mock):
+        action = ActionsExample()
+        action.first_step()
+        action.second_step()
+
+        assert "notice:: Step 1 running!" in annotated_logger_mock.messages[0]
+        annotated_logger_mock.assert_logged("DEBUG", count=0)