为何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
相关产品推荐
相关产品推荐

