You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

为何Python logging中处理器名称以`s`开头会触发报错?

问题现象

为asyncio应用配置logging时,使用dictConfig结合QueueHandler规避阻塞I/O问题:

  • 当处理器名为console时,运行main.py可正常输出HELLO WORLD
  • 但将处理器名改为s或以s开头的名称(如stream、stdout)时,会抛出错误:
    AttributeError: 'ConvertingDict' object has no attribute 'handle'
    

该问题在Python 3.9-3.11版本、MacOS与Ubuntu系统下均可复现。进一步排查发现,logging.config._resolve_handlers函数对非s开头的处理器返回实例,对s开头的返回配置字典(ConvertingDict),导致QueueListener调用handle方法失败。

根因分析

Python 3.9-3.11版本的logging.config模块中,_resolve_handlers函数存在逻辑缺陷:当解析QueueHandler配置中的handlers参数时,若处理器名称以s开头,函数会错误地返回该处理器的原始配置字典(ConvertingDict),而非已经实例化的处理器对象。而QueueListener期望接收的是具备handle方法的处理器实例,因此触发属性错误。

解决方案

方案1:规避使用s开头的处理器名称

这是最直接的解决方式,将处理器名称改为非s开头的字符串,比如console、app_stream等,示例配置如下:

import logging.config
import asyncio
from logging.handlers import QueueListener
import queue

def setup_logging():
    log_queue = queue.Queue(-1)
    config = {
        "version": 1,
        "disable_existing_loggers": False,
        "handlers": {
            "console": {  # 非s开头的名称
                "class": "logging.StreamHandler",
                "formatter": "simple",
            },
            "queue_handler": {
                "class": "logging.handlers.QueueHandler",
                "handlers": ["console"],
                "queue": log_queue,
            },
        },
        "formatters": {
            "simple": {"format": "%(levelname)s: %(message)s"},
        },
        "root": {"level": "INFO", "handlers": ["queue_handler"]},
    }
    logging.config.dictConfig(config)
    # 启动QueueListener
    listener = QueueListener(log_queue, *logging.getLogger().handlers[0].handlers)
    listener.start()
    return listener

async def main():
    logging.getLogger(__name__).info("HELLO WORLD")

if __name__ == "__main__":
    listener = setup_logging()
    try:
        asyncio.run(main())
    finally:
        listener.stop()

方案2:手动实例化处理器(适合必须使用s开头名称的场景)

绕过dictConfig的自动解析逻辑,手动创建处理器实例并直接传入QueueHandler和QueueListener:

import logging
import asyncio
from logging.handlers import QueueHandler, QueueListener
import queue

def setup_logging():
    log_queue = queue.Queue(-1)
    # 手动创建s开头的处理器实例
    stream_handler = logging.StreamHandler()
    stream_handler.setFormatter(logging.Formatter("%(levelname)s: %(message)s"))
    
    # 配置QueueHandler与根日志器
    queue_handler = QueueHandler(log_queue)
    root_logger = logging.getLogger()
    root_logger.addHandler(queue_handler)
    root_logger.setLevel(logging.INFO)
    
    # 启动QueueListener
    listener = QueueListener(log_queue, stream_handler)
    listener.start()
    return listener

async def main():
    logging.getLogger(__name__).info("HELLO WORLD")

if __name__ == "__main__":
    listener = setup_logging()
    try:
        asyncio.run(main())
    finally:
        listener.stop()

内容的提问来源于stack exchange,提问作者Zion

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.03 13:01:04