如何用 structlog 和 stdlib logging 输出扁平 JSON？

2026-11-04数据小许

结论

使用 stdlib logging 输出扁平 JSON 时，建议由 structlog.stdlib.ProcessorFormatter 负责最终格式化，并将 structlog.stdlib.ProcessorFormatter.wrap_for_formatter 放在 structlog processor 链的末尾。事件字段直接作为关键字参数传给 structlog，例如 logger.warning("check_failed", user_id=222, reason="timeout")。

这条处理链中不要混用 render_to_log_kwargs、自定义 extra 展开和 JSONRenderer。extra 用于设置 stdlib logging 的 LogRecord 属性，不是通用的 structlog 事件字段容器。

推荐配置

import logging
import structlog


def configure_events(handler: logging.Handler) -> None:
    shared_processors = [
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(
            fmt="%Y-%m-%d %H:%M:%S",
            utc=False,
        ),
    ]

    structlog.configure(
        processors=[
            *shared_processors,
            structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
        ],
        logger_factory=structlog.stdlib.LoggerFactory(),
        wrapper_class=structlog.stdlib.BoundLogger,
        cache_logger_on_first_use=True,
    )

    handler.setFormatter(
        structlog.stdlib.ProcessorFormatter(
            foreign_pre_chain=shared_processors,
            processors=[
                structlog.stdlib.ProcessorFormatter.remove_processors_meta,
                structlog.processors.JSONRenderer(),
            ],
        )
    )

    root_logger = logging.getLogger()
    root_logger.addHandler(handler)
    root_logger.setLevel(logging.INFO)

配置好后，直接传入事件字段：

logger = structlog.get_logger()
logger.warning("check_failed", user_id=222, reason="timeout")

最终 JSON 会将 event 作为事件文本输出，并把 level、时间戳、user_id、reason 等字段放在同一层级。事件文本对应的 JSON 键名取决于所用 renderer 及其版本行为。如果键名必须是 msg，应在最终输出时明确映射事件字段，并通过实际序列化结果验证。

为什么会出现两个异常

logger.warning() 会以 WARNING 级别调用 stdlib logging。报错 Logger._log() got an unexpected keyword argument 'user_id' 说明 user_id 被传到了 stdlib 日志方法，而没有进入预期的 structlog 事件处理和格式化流程。使用标准库集成时，wrap_for_formatter 和 ProcessorFormatter 会将结构化事件交给 handler 格式化，无需把事件字段改成 extra。

Logger._log() got multiple values for argument 'level' 说明调用过程中，日志级别既通过日志方法传入，也作为关键字参数传入。stdlib 的 warning() 已经确定了级别；如果把名为 level 的事件字段展开成传给 stdlib 的顶层关键字参数，就会发生冲突。INFO 和 WARNING 方法使用事件字段的方式相同。出现这个差异，说明当前 processor 链生成的调用参数不合适，不代表应该为 WARNING 单独传 extra。

测试最终 JSON

可以捕获 handler 写出的内容，再解析 JSON。这样测试验证的是最终输出，而不只是中间 processor 生成的事件字典：

import io
import json
import logging


def test_event_can_use_warning_level():
    stream = io.StringIO()
    handler = logging.StreamHandler(stream)
    configure_events(handler)

    logger = structlog.get_logger()
    logger.warning("check_failed", user_id=222, reason="timeout")

    log = json.loads(stream.getvalue())
    assert log["level"] == "warning"
    assert log["user_id"] == 222
    assert log["reason"] == "timeout"

重复运行测试时，应隔离 logging handler，避免根 logger 上累积多个 handler。每个测试也应先完成 structlog 配置，再创建或首次使用 logger，因为配置启用了 cache_logger_on_first_use。

注意事项

structlog 事件的关键字参数用于结构化事件字段；stdlib logging 的 extra 用于填充 LogRecord 属性。
render_to_log_kwargs 也是一种将 structlog 事件交给 stdlib logging 的方式。使用 ProcessorFormatter 时，不要再把它加入同一条处理链。
如果还要兼容其他模块直接使用 stdlib logging，可以通过 ProcessorFormatter 格式化这些日志；foreign_pre_chain 用于为它们补充共享处理器。
上述配置会把根 logger 的最低级别设为 INFO。如果应用已有 handler 或级别设置，应根据现有日志配置调整，避免重复添加 handler。