Python Structlog JSON

Question

I am currently using the python structlog JSONRenderer and hoping to change my log configuration to render the event as the 1st JSON attribute for better readability.

Current Configuration:

structlog.configure(processors=[structlog.processors.JSONRenderer()])
log = structlog.get_logger()

Current log call site:

log.msg("Response: ",
                        content_type=content_type,
                        content_length=resp.headers.get('content-length'),
                        status_code=resp.status_code
                )

Current Output:

{"content_type": "application/json", "content_length": null, "status_code": 200, "event": "Response: "}

Desired Output:

{"event": "Response: ", "content_type": "application/json", "content_length": null, "status_code": 200}

Any assistance would be greatly appreciated.

Answer 1

The structlog.processors.JSONRenderer just passes the log object to json.dumps unless you specify another callable instead:

structlog.configure(processors=[structlog.processors.JSONRenderer(serializer=mydumps)])

The mydumps will then be a function that does what json.dumps does but puts event first. This could look like:

def mydumps(dic,**kw):
   mod = {}
   if 'event' in dic:
      mod["event"] = dic["event"]
   for k in dic:
      if k!="event":
         mod[k] = dic[k]
   return json.dumps(mod,**kw)

What it does is to make a new object then look for event key in the input object and put it first to the new object then proceeding to put rest of keys into the object and pass it along with **kw to json.dumps.

Note that this way you would not need to specify beforehand what other keys your logs might have (like content-type) as any event type might have different info.

Answer 2

It looks like you might be using an older version of python than 3.6 which keeps keys ordered in insertion order. You can use the KeyValueRenderer to set the key order and use OrderedDict as the context_class:

from collections import OrderedDict

structlog.configure(
    processors=[
        structlog.processors.KeyValueRenderer(
            key_order=["event", "content_type", "content_length", "status_code"]
        ),
        structlog.processors.JSONRenderer()
    ],
    context_class = OrderedDict
)
log = structlog.get_logger()

Reference: KeyValueRenderer

Answer 3

I merged and improved upon both answers by re-using what is already in a structlog.

The approved answer used a custom json dump method that I want to avoid, the other solution has the side-effect of converting log from JSON to string, losing a lot of usefulness.

Complete example of mine that works both in local shell and remote server would be:

import collections
import logging
import sys

import orjson
import structlog
from structlog.typing import EventDict, WrappedLogger


class ForcedKeyOrderRenderer(structlog.processors.KeyValueRenderer):
    """Based upon KeyValueRenderer but returns dict instead of string."""

    def __call__(self, _: WrappedLogger, __: str, event_dict: EventDict) -> str:
        sorted_dict = self._ordered_items(event_dict)
        return collections.OrderedDict(**{key: value for key, value in sorted_dict})


shared_processors = [
    structlog.contextvars.merge_contextvars,
    structlog.processors.add_log_level,
    structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S", utc=True),
]

if sys.stderr.isatty():
    processors = shared_processors + [
        structlog.processors.StackInfoRenderer(),
        structlog.dev.set_exc_info,
        structlog.dev.ConsoleRenderer(),
    ]
    logger_factory = None
else:  # pragma: no cover
    processors = shared_processors + [
        structlog.processors.format_exc_info,
        structlog.processors.dict_tracebacks,
        ForcedKeyOrderRenderer(
            sort_keys=True,
            key_order=[
                "event",
                "content_type",
                "content_length",
                "status_code",
            ],
            drop_missing=True,
        ),
        structlog.processors.JSONRenderer(serializer=orjson.dumps),
    ]
    logger_factory = structlog.BytesLoggerFactory()

if not structlog.is_configured():
    structlog.configure(
        cache_logger_on_first_use=True,
        wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
        processors=processors,
        logger_factory=logger_factory,
    )

logger = structlog.get_logger()