feat(log): colorful logs for stdlib logging (#242)

* ✨ feat(log): colorful logs for stdlib logging

* ♻️ refactor: remove commented code

* 🐛 fix(log): json logging

* 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.

Co-authored-by: Peter Schutt <[email protected]>

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
@@ -35,12 +35,23 @@
     structlog.processors.TimeStamper(fmt="iso", utc=True),
 ]
 
+stdlib_processors = [
+    structlog.processors.TimeStamper(fmt="iso", utc=True),
+    structlog.stdlib.add_log_level,
+    structlog.stdlib.ExtraAdder(),
+    EventFilter(["color_message"]),
+    structlog.stdlib.ProcessorFormatter.remove_processors_meta,
+]
+
 if settings.app.ENVIRONMENT == "local":  # pragma: no cover
     LoggerFactory: Any = structlog.WriteLoggerFactory
-    default_processors.extend([structlog.dev.ConsoleRenderer()])
+    console_processor = structlog.dev.ConsoleRenderer(colors=True)
+    default_processors.extend([console_processor])
+    stdlib_processors.append(console_processor)
 else:
     LoggerFactory = structlog.BytesLoggerFactory
     default_processors.extend([structlog.processors.dict_tracebacks, msgspec_json_renderer])
+    stdlib_processors.append(structlog.processors.dict_tracebacks)
 
 
 def configure(processors: Sequence[Processor]) -> None:
@@ -62,12 +73,7 @@ def configure(processors: Sequence[Processor]) -> None:
 config = LoggingConfig(
     root={"level": logging.getLevelName(settings.log.LEVEL), "handlers": ["queue_listener"]},
     formatters={
-        "standard": {
-            "format": (
-                "%(asctime)s loglevel=%(levelname)-6s logger=%(name)s %(funcName)s() "
-                "L%(lineno)-4d %(message)s"
-            )
-        }
+        "standard": {"()": structlog.stdlib.ProcessorFormatter, "processors": stdlib_processors}
     },
     loggers={
         "uvicorn.access": {

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"}