structlog_handler_usage.py

import logging

import structlog
from structlog.contextvars import bound_contextvars
from structlog.processors import CallsiteParameter as CsParam
from structlog.typing import FilteringBoundLogger

from structlog_stdlib_extras import StructlogHandler


def configure():
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso", utc=True),
            structlog.processors.StackInfoRenderer(),
            structlog.processors.CallsiteParameterAdder(
                {CsParam.FILENAME, CsParam.LINENO, CsParam.MODULE, CsParam.FUNC_NAME}
            ),
            structlog.dev.ConsoleRenderer(),
        ],
        wrapper_class=structlog.make_filtering_bound_logger(logging.DEBUG),
        logger_factory=structlog.WriteLoggerFactory(),
        cache_logger_on_first_use=True,
    )

    root_logger = logging.getLogger()
    root_logger.addHandler(StructlogHandler())
    root_logger.setLevel(logging.DEBUG)


def main():
    logging.getLogger("our.app").info("Hello, World!")

    structured_logger: FilteringBoundLogger = structlog.get_logger("our.app")
    structured_logger.info("Hello from structlog!", some_key="some_value")

    with bound_contextvars(another_key="another_value"):
        # Context vars are merged even when using the standard logging module
        logging.getLogger("our.app").info("Hello again!")


if __name__ == "__main__":
    configure()
    main()

structlog_handler_usage_json.py

import logging

import orjson
import structlog
from structlog.contextvars import bound_contextvars
from structlog.processors import CallsiteParameter as CsParam
from structlog.typing import FilteringBoundLogger

from structlog_stdlib_extras import StructlogHandler


def configure():
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso", utc=True),
            structlog.processors.StackInfoRenderer(),
            structlog.processors.CallsiteParameterAdder(
                {CsParam.FILENAME, CsParam.LINENO, CsParam.MODULE, CsParam.FUNC_NAME}
            ),
            structlog.processors.JSONRenderer(orjson.dumps),
        ],
        wrapper_class=structlog.make_filtering_bound_logger(logging.DEBUG),
        logger_factory=structlog.BytesLoggerFactory(),
        cache_logger_on_first_use=True,
    )

    root_logger = logging.getLogger()
    root_logger.addHandler(StructlogHandler())
    root_logger.setLevel(logging.DEBUG)


def main():
    logging.getLogger("our.app").info("Hello, World!")

    structured_logger: FilteringBoundLogger = structlog.get_logger("our.app")
    structured_logger.info("Hello from structlog!", some_key="some_value")

    with bound_contextvars(another_key="another_value"):
        # Context vars are merged even when using the standard logging module
        logging.getLogger("our.app").info("Hello again!")


if __name__ == "__main__":
    configure()
    main()

structlog_stdlib_extras.py

import logging
from collections.abc import Collection
from typing import final, cast

import structlog
from structlog.typing import Processor, EventDict


@final
class StructlogHandler(logging.Handler):
    def __init__(
        self,
        pre_chain: list[Processor] | None = None,
        *,
        use_get_message: bool = True,
        pass_foreign_args: bool = False,
        level: int = logging.NOTSET,
    ):
        super().__init__(level)
        self.processors: Collection[Processor] = pre_chain or [
            structlog.stdlib.add_logger_name,
            structlog.stdlib.ExtraAdder(),
            structlog.stdlib.ProcessorFormatter.remove_processors_meta,
        ]
        self.use_get_message = use_get_message
        self.pass_foreign_args = pass_foreign_args
        self._logger = structlog.get_logger()

    def format(self, record: logging.LogRecord) -> str:
        if formatter := self.formatter:
            return formatter.format(record)
        return record.getMessage() if self.use_get_message else str(record.msg)

    def process(self, record: logging.LogRecord) -> EventDict:
        logger = None
        meth_name = record.levelname.lower()
        ed: EventDict = {
            "event": self.format(record),
            "_record": record,
            "_from_structlog": False,
        }

        if self.pass_foreign_args:
            ed["positional_args"] = record.args

        # Add stack-related attributes to the event dict
        if record.exc_info:
            ed["exc_info"] = record.exc_info
        if record.stack_info:
            ed["stack_info"] = record.stack_info

        for proc in self.processors or ():
            ed = cast(EventDict, proc(logger, meth_name, ed))

        return ed

    def handle(self, record: logging.LogRecord) -> None:
        if self.level > record.levelno:
            return
        super().handle(record)

    def emit(self, record: logging.LogRecord) -> None:
        try:
            event_dict = self.process(record)
            event: str = event_dict.pop("event")
            self._logger.log(record.levelno, event, **event_dict)
        except Exception:  # noqa
            self.handleError(record)

    def flush(self) -> None:
        if flush := getattr(self._logger, "flush", None):
            flush()