StructlogFormatter conforms to structlog.typing.Processor

Closes #146

Author: AWhetter

ecs_logging/_structlog.py

@@ -17,7 +17,13 @@
 
 import time
 import datetime
-from typing import Any, Dict
+import sys
+from typing import Any
+
+if sys.version_info >= (3, 9):
+    from collections.abc import MutableMapping
+else:
+    from typing import MutableMapping
 
 from ._meta import ECS_VERSION
 from ._utils import json_dumps, normalize_dict
@@ -26,7 +32,7 @@
 class StructlogFormatter:
     """ECS formatter for the ``structlog`` module"""
 
-    def __call__(self, _: Any, name: str, event_dict: Dict[str, Any]) -> str:
+    def __call__(self, _: Any, name: str, event_dict: MutableMapping[str, Any]) -> str:
 
         # Handle event -> message now so that stuff like `event.dataset` doesn't
         # cause problems down the line
@@ -36,7 +42,9 @@ def __call__(self, _: Any, name: str, event_dict: Dict[str, Any]) -> str:
         event_dict = self.format_to_ecs(event_dict)
         return self._json_dumps(event_dict)
 
-    def format_to_ecs(self, event_dict: Dict[str, Any]) -> Dict[str, Any]:
+    def format_to_ecs(
+        self, event_dict: MutableMapping[str, Any]
+    ) -> MutableMapping[str, Any]:
         if "@timestamp" not in event_dict:
             event_dict["@timestamp"] = (
                 datetime.datetime.fromtimestamp(
@@ -55,5 +63,5 @@ def format_to_ecs(self, event_dict: Dict[str, Any]) -> Dict[str, Any]:
         event_dict.setdefault("ecs.version", ECS_VERSION)
         return event_dict
 
-    def _json_dumps(self, value: Dict[str, Any]) -> str:
+    def _json_dumps(self, value: MutableMapping[str, Any]) -> str:
         return json_dumps(value=value)

ecs_logging/_utils.py

@@ -18,8 +18,14 @@
 import collections.abc
 import json
 import functools
+import sys
 from typing import Any, Dict, Mapping
 
+if sys.version_info >= (3, 9):
+    from collections.abc import MutableMapping
+else:
+    from typing import MutableMapping
+
 
 __all__ = [
     "normalize_dict",
@@ -53,9 +59,9 @@ def flatten_dict(value: Mapping[str, Any]) -> Dict[str, Any]:
     return top_level
 
 
-def normalize_dict(value: Dict[str, Any]) -> Dict[str, Any]:
+def normalize_dict(value: MutableMapping[str, Any]) -> MutableMapping[str, Any]:
     """Expands all dotted names to nested dictionaries"""
-    if not isinstance(value, dict):
+    if not isinstance(value, MutableMapping):
         return value
     keys = list(value.keys())
     for key in keys:
@@ -78,7 +84,9 @@ def de_dot(dot_string: str, msg: Any) -> Dict[str, Any]:
     return ret
 
 
-def merge_dicts(from_: Dict[Any, Any], into: Dict[Any, Any]) -> Dict[Any, Any]:
+def merge_dicts(
+    from_: Mapping[Any, Any], into: MutableMapping[Any, Any]
+) -> MutableMapping[Any, Any]:
     """Merge deeply nested dictionary structures.
     When called has side-effects within 'destination'.
     """
@@ -98,7 +106,7 @@ def merge_dicts(from_: Dict[Any, Any], into: Dict[Any, Any]) -> Dict[Any, Any]:
     return into
 
 
-def json_dumps(value: Dict[str, Any]) -> str:
+def json_dumps(value: MutableMapping[str, Any]) -> str:
 
     # Ensure that the first three fields are '@timestamp',
     # 'log.level', and 'message' per ECS spec

pyproject.toml

@@ -33,6 +33,7 @@ develop = [
     "mock",
     "structlog",
     "elastic-apm",
+    "mypy",
 ]
 
 [tool.flit.metadata.urls]

tests/test_typing.py

@@ -0,0 +1,14 @@
+import pathlib
+import subprocess
+import sys
+
+import pytest
+
+_THIS_DIR = pathlib.Path(__file__).parent
+
+
+@pytest.mark.parametrize("file", (_THIS_DIR / "typing").glob("*.py"))
+def test_type_check(file):
+    subprocess.check_call(
+        [sys.executable, "-m", "mypy", "--strict", "--config-file=", file]
+    )

tests/typing/structlog_usage.py

@@ -0,0 +1,34 @@
+import sys
+from typing import List, Tuple
+
+import ecs_logging
+import structlog
+from structlog.typing import Processor
+
+
+shared_processors: Tuple[Processor, ...] = (
+    structlog.contextvars.merge_contextvars,
+    structlog.processors.add_log_level,
+    structlog.processors.StackInfoRenderer(),
+    structlog.dev.set_exc_info,
+    structlog.processors.TimeStamper(fmt="iso", utc=True),
+)
+
+processors: List[Processor]
+if sys.stderr.isatty():
+    processors = [
+        *shared_processors,
+        structlog.dev.ConsoleRenderer(),
+    ]
+else:
+    processors = [
+        *shared_processors,
+        structlog.processors.dict_tracebacks,
+        ecs_logging.StructlogFormatter(),
+    ]
+
+structlog.configure(
+    processors=processors,
+    logger_factory=structlog.PrintLoggerFactory(),
+    cache_logger_on_first_use=True,
+)