structlog: display log level in Cloudwatch

Question

I have set up my logger like this:

import logging
import structlog


class Logging:
    @staticmethod
    def my_logger() -> logging.Logger:
        structlog.configure(
            processors=[
                structlog.processors.add_log_level,
                structlog.processors.TimeStamper(fmt="iso", key="ts"),
                structlog.processors.JSONRenderer(),
            ],
            wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
        )
        logger = structlog.getLogger()
        return logger

Events in Cloudwatch now look like this:

2023-04-05T10:44:52.920+01:00   {"event": "My logging message", "level": "info", "ts": "2023-04-05T09:44:52.919867Z"}

Instead, I want to see the log level right at the beginning like I would with the default logging module:

2023-04-05T10:44:52.920+01:00   [INFO] {"event": "My logging message", "level": "info", "ts": "2023-04-05T09:44:52.919867Z"}

How can I accomplish this?

Answer 1

The simplest way is to prepend the method name before the rendered JSON:

import logging
import structlog


def prepend_level(_, method: str, line: str) -> str:
    return f"[{method.upper()}] {line}"


class Logging:
    @staticmethod
    def my_logger() -> logging.Logger:
        structlog.configure(
            processors=[
                structlog.processors.add_log_level,
                structlog.processors.TimeStamper(fmt="iso", key="ts"),
                structlog.processors.JSONRenderer(),
                prepend_level,
            ],
            wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
        )
        logger = structlog.getLogger()
        return logger


Logging.my_logger().info("hey")

Gives me:

[INFO] {"event": "hey", "level": "info", "ts": "2023-04-05T12:19:33.803902Z"}

The cleaner version is wrapping the json renderer:

import logging
import structlog


class Logging:
    @staticmethod
    def my_logger() -> logging.Logger:
        json_renderer = structlog.processors.JSONRenderer()

        def render(logger, method, ed):
            return f"[{ed['level'].upper()}] {json_renderer(logger, method, ed)}"

        structlog.configure(
            processors=[
                structlog.processors.add_log_level,
                structlog.processors.TimeStamper(fmt="iso", key="ts"),
                render,
            ],
            wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
        )
        logger = structlog.getLogger()
        return logger


Logging.my_logger().info("hey")