Merge pull request #14 from vipyrsec/additional_processors_parameter

Add additional_processors parameter to `configure_logger`.

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "dragonfly-logging-config"
-version = "1.0.0"
+version = "1.1.0"
 description = "Shared logging configuration for Dragonfly codebases"
 authors = [{ name = "Vipyr Security", email = "[email protected]" }]
 dependencies = [

src/logging_config/__init__.py

@@ -1,10 +1,12 @@
 import logging
 import logging.config
+from typing import Any
+from typing import Optional
 
 import structlog
 
 
-def configure_logger(config: dict):
+def configure_logger(config: dict[str, Any], additional_processors: Optional[list[Any]] = None):
     # Define the shared processors, regardless of whether API is running in prod or dev.
     shared_processors: list[structlog.types.Processor] = [
         structlog.contextvars.merge_contextvars,
@@ -24,6 +26,9 @@ def configure_logger(config: dict):
         ),
     ]
 
+    if additional_processors:
+        shared_processors.extend(additional_processors)
+
     config.update(
         {
             "version": 1,

src/tests.py

@@ -1,6 +1,8 @@
 import io
+import logging
 import structlog
 import unittest
+from typing import Any
 
 from logging_config import configure_logger
 
@@ -50,3 +52,42 @@ def test_log_output(self):
                 "func_name": "test_log_output",
             },
         )
+
+    def test_additional_processors(self):
+        """
+        Test whether passing additional processors works correctly.
+        """
+
+        f = io.StringIO()
+
+        config = {
+            "disable_existing_loggers": False,
+            "handlers": {
+                "default": {
+                    "level": "DEBUG",
+                    "class": "logging.StreamHandler",
+                    "stream": f,
+                    "formatter": "json",
+                },
+            },
+            "loggers": {
+                "": {"handlers": ["default"], "level": "DEBUG", "propagate": True},
+            },
+        }
+
+        def additional_processor(
+            logger: logging.Logger, method_name: str, event_dict: dict[str, Any]
+        ) -> dict[str, Any]:
+            event_dict["testing"] = "testing"
+            return event_dict
+
+        configure_logger(config, additional_processors=[additional_processor])
+
+        cf = structlog.testing.CapturingLoggerFactory()
+        structlog.configure(logger_factory=cf)
+        log = structlog.get_logger()
+
+        log.info("test")
+        record = cf.logger.calls[0]
+
+        self.assertEqual(record.args[0]["testing"], "testing")