feat(log): EventFilter processor type.

A processor class that allows specific keys to be filtered from a log
event.

Applies the filter to our default processor chain to remove the
"color_message" key that is injected into extras by uvicorn.

src/starlite_saqlalchemy/log/__init__.py

@@ -11,7 +11,7 @@
 from starlite_saqlalchemy import settings
 
 from . import controller, worker
-from .msgspec_renderer import msgspec_json_renderer
+from .utils import EventFilter, msgspec_json_renderer
 
 if TYPE_CHECKING:
     from collections.abc import Sequence
@@ -39,6 +39,7 @@
     structlog.processors.TimeStamper(fmt="iso", utc=True),
     structlog.stdlib.add_log_level,
     structlog.stdlib.ExtraAdder(),
+    EventFilter(["color_message"]),
     structlog.stdlib.ProcessorFormatter.remove_processors_meta,
 ]
 

src/starlite_saqlalchemy/log/utils.py

@@ -0,0 +1,79 @@
+"""Logging utilities.
+
+`msgspec_json_renderer()`
+    A JSON Renderer for structlog using msgspec.
+
+    Msgspec doesn't have an API consistent with the stdlib's `json` module,
+    which is required for structlog's `JSONRenderer`.
+
+`EventFilter`
+    A structlog processor that removes keys from the log event if they exist.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import msgspec
+
+if TYPE_CHECKING:
+    from collections.abc import Iterable
+
+    from structlog.typing import EventDict, WrappedLogger
+
+
+_encoder = msgspec.json.Encoder()
+
+
+def msgspec_json_renderer(_: WrappedLogger, __: str, event_dict: EventDict) -> bytes:
+    """Structlog processor that uses `msgspec` for JSON encoding.
+
+    Args:
+        _ ():
+        __ ():
+        event_dict (): The data to be logged.
+
+    Returns:
+        The log event encoded to JSON by msgspec.
+    """
+    return _encoder.encode(event_dict)
+
+
+class EventFilter:
+    """Remove keys from the log event.
+
+    Add an instance to the processor chain.
+
+    Examples
+        structlog.configure(
+            ...,
+            processors=[
+                ...,
+                EventFilter(["color_message"]),
+                ...,
+            ]
+        )
+    """
+
+    def __init__(self, filter_keys: Iterable[str]) -> None:
+        """
+
+        Args:
+            filter_keys: Iterable of string keys to be excluded from the log event.
+        """
+        self.filter_keys = filter_keys
+
+    def __call__(self, _: WrappedLogger, __: str, event_dict: EventDict) -> EventDict:
+        """Receive the log event, and filter keys.
+
+        Args:
+            _ ():
+            __ ():
+            event_dict (): The data to be logged.
+
+        Returns:
+            The log event with any key in `self.filter_keys` removed.
+        """
+        for key in self.filter_keys:
+            event_dict.pop(key, None)
+        return event_dict

tests/unit/test_log.py

@@ -452,3 +452,11 @@ def test_handler(data: dict[str, Any]) -> dict[str, Any]:
     client.app.register(test_handler)
     resp = client.post("/", content=b'{"a": "b",}', headers={"content-type": "application/json"})
     assert resp.status_code == HTTP_400_BAD_REQUEST
+
+
+def test_event_filter() -> None:
+    """Functionality test for the event filter processor."""
+    event_filter = log.utils.EventFilter(["a_key"])
+    log_event = {"a_key": "a_val", "b_key": "b_val"}
+    log_event = event_filter(..., "", log_event)  # type:ignore[assignment]
+    assert log_event == {"b_key": "b_val"}