使用structlog结合标准库logging实现扁平JSON日志时，WARNING级别日志参数传递报错的问题

嗨，我来帮你搞定这个头疼的问题！你遇到的核心矛盾是structlog和标准库logging的参数传递逻辑不兼容，再加上处理器顺序没配对，才导致INFO级别正常、WARNING级别报错的奇怪情况。

问题根源拆解 #

你之前的配置里用了structlog.stdlib.render_to_log_kwargs这个处理器，它会把structlog的事件字典打包成extra参数传给标准库logging。但标准库logging的不同级别方法（比如warning）在底层调用_log时，会自动把日志级别作为参数传递；而render_to_log_kwargs生成的参数会和这个自动传递的参数冲突，再加上你手动处理extra的逻辑，就出现了要么参数不识别、要么参数重复的错误。

另外，你自定义的_flatten_extra处理器位置不对，放在render_to_log_kwargs之后时，参数已经被转换成logging的格式，这时候修改event_dict反而会干扰logging的参数解析逻辑。

正确的配置方案 #

我们换用structlog官方推荐的与标准库logging集成的方式，用ProcessorFormatter来衔接两者，既能保证日志是扁平的JSON结构，又不会出现参数冲突：

def configure_events() -> None:
    # structlog侧的处理器链，负责预处理事件
    processors = [
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S", utc=False),
        structlog.stdlib.PositionalArgumentsFormatter(),
        # 关键：用这个处理器把事件打包成logging能识别的格式
        structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
    ]

    structlog.configure(
        processors=processors,
        wrapper_class=structlog.make_filtering_bound_logger("INFO"),
        logger_factory=structlog.stdlib.LoggerFactory(),
        cache_logger_on_first_use=True,
    )

    # 配置标准库logging的处理器，负责最终的JSON渲染
    import logging
    handler = logging.StreamHandler()
    handler.setFormatter(structlog.stdlib.ProcessorFormatter(
        # 最终输出扁平JSON
        processor=structlog.processors.JSONRenderer(),
        # 如果需要额外预处理，在这里加前置处理器
        foreign_pre_chain=[],
    ))
    
    root_logger = logging.getLogger()
    root_logger.addHandler(handler)
    root_logger.setLevel(logging.INFO)

调整后的测试代码 #

现在不管是INFO还是WARNING级别，都可以直接传关键字参数，不需要手动打包extra：

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"
    assert capture_logs[0]["user_id"] == 222
    assert capture_logs[0]["reason"] == "timeout"

同时，我们可以把测试的日志捕获逻辑改成直接监听标准库logging的输出，这样更准确：

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

    class CaptureHandler(logging.Handler):
        def emit(self, record):
            # 直接获取ProcessorFormatter处理后的事件字典
            emitted.append(record.msg)

    handler = CaptureHandler()
    root_logger = logging.getLogger()
    root_logger.addHandler(handler)
    yield emitted
    root_logger.removeHandler(handler)

为什么这样能解决问题？ #

structlog.stdlib.ProcessorFormatter.wrap_for_formatter会正确处理structlog事件到标准库logging参数的转换，避免了手动处理extra带来的冲突；而最终的JSONRenderer会直接把所有字段（包括级别、时间戳、自定义字段）渲染成扁平的JSON结构，完全符合你的需求。

备注：内容来源于stack exchange，提问作者novice