我想用Python的 structlog 实现全面的日志记录。我也想为它提供一些单元测试，并且我很困惑如何把 structlog 与底层标准库 logging. 搭配使用。总体来说，我想实现的是将所有日志格式化成一个简单的JSON，且没有嵌套的键，例如：

{
    "msg": "some event",
    "level": "warning",
    "timestamp": "2026-03-09 19:01:20",
    "user_id": 123,
    "frequency": "some key"
    // some other key-value pairs
}

初始 structlog 配置：

def configure_events() -> None:
    processors = [
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S", utc=False),
        structlog.stdlib.render_to_log_kwargs,
        structlog.stdlib.PositionalArgumentsFormatter(),
        structlog.processors.JSONRenderer()
    ]
    structlog.configure(
        processors=processors,
        wrapper_class=structlog.make_filtering_bound_logger("INFO"),
        logger_factory=structlog.stdlib.LoggerFactory(),
        cache_logger_on_first_use=True,
    )

测试配置

@pytest.fixture
def capture_logs():
    emitted = []

    def capture(*args):
        event_dict = args[-1]
        emitted.append(event_dict)
        return event_dict

    with patch("structlog.processors.JSONRenderer.__call__", capture):
        yield emitted

当我像这样运行测试时：

def test_event_emits_structured_wide_event(capture_logs):
    configure_events()
    logger = structlog.get_logger()

    logger.info("some event", user_id=123)

    assert len(capture_logs) == 1
    log = capture_logs[0]

    assert log["msg"] == "some event"
    assert log["extra"]["level"] == "info"
    assert log["extra"]["user_id"] == 123

测试运行得很好，但 level 和 user_id 位于 extra 键下，这可以理解，因为日志会按传入的参数来处理。可是我不想要这样，我希望所有键都只是一个简单的JSON。我尝试了很多不同的方法，最终我想到了添加一个自定义处理器，将 extra 键的值扁平化，因此实现之后，我的 structlog 配置看起来是这样的：

def configure_events() -> None:

    def _flatten_extra(logger, method_name, event_dict):
        if "extra" in event_dict:
            extra = event_dict.pop("extra")
            if not isinstance(extra, dict):
                raise ValueError("Value of 'extra' is not of dict type.")
            event_dict.update(extra)
        return event_dict

    processors = [
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S", utc=False),
        structlog.stdlib.render_to_log_kwargs,
        structlog.stdlib.PositionalArgumentsFormatter(),
        _flatten_extra,
        structlog.processors.JSONRenderer()
    ]
    structlog.configure(
        processors=processors,
        wrapper_class=structlog.make_filtering_bound_logger("INFO"),
        logger_factory=structlog.stdlib.LoggerFactory(),
        cache_logger_on_first_use=True,
    )

然后在我的断言中移除了对 extra 的引用后，测试就顺利通过。

但这在WARNING上不起作用。 我有如下测试：

def test_event_can_use_warning_level(capture_logs):
    configure_events()
    logger = structlog.get_logger()
    logger.warning("check_failed", user_id=222, reason="timeout")
    assert capture_logs[0]["level"] == "warning"

测试因为 TypeError: Logger._log() got an unexpected keyword argument 'user_id' 而失败，这点对我来说很奇怪，为什么不能像INFO那样传参，但也无妨，下面是我测试的内容：

def test_event_can_use_warning_level(capture_logs):
    configure_events()
    logger = structlog.get_logger()
    extra = {"user_id": 222, "reason": "timeout"}
    logger.warning("check_failed", extra=extra)
    assert capture_logs[0]["level"] == "warning"

而这次测试的结果让我陷入了主要的困惑，它因为 TypeError: Logger._log() got multiple values for argument 'level' 而失败。根据我的理解，标准库 logging 会在日志消息中添加一个由方法名派生的 level 关键字，但为什么它在INFO和 WARNING之间的行为会不同？

我现在完全搞不清楚，也想不出办法。我到底错过了什么，还是配置错了？

解决方案

你在 JSONRenderer.__call__ 的mock函数中返回了一个 dict，这将导致未定义行为。你可以按下面的示例来修复。

@pytest.fixture
def capture_logs():
    emitted = []

    old_call = structlog.processors.JSONRenderer.__call__
    def capture(*args):
        event_dict = args[-1]
        emitted.append(event_dict)
        return old_call(*args)

    with patch("structlog.processors.JSONRenderer.__call__", capture):
        yield emitted