如何将Python logging.Logger实例作为流传入invoke的out_stream参数？

问题背景 #

我正在使用invoke模块的Runner.run方法运行/usr/bin/printf程序。该方法的out_stream参数接受类文件流对象（默认使用sys.stdout），用于接收子进程的标准输出。

当前运行命令的代码如下：

import invoke

def print_foobar(context: invoke.context):
    context.run("/usr/bin/printf foobar")

我希望将logging.Logger实例传入out_stream，让子进程printf的每一行输出都生成一条log.info级别的日志，类似如下伪代码的效果：

import invoke
import logging

log = logging.getLogger()

def print_foobar(context: invoke.context):
    context.run("/usr/bin/printf foobar", out_stream=log.stream)

核心需求：如何把Logger实例包装成符合要求的类文件流，使其对接收到的每一行文本生成对应日志消息？

解决方案 #

你可以自定义一个类文件流包装器，实现必要的write和flush方法，将写入的内容转发给Logger实例。

1. 实现日志流包装类 #

import logging

class LogStream:
    def __init__(self, logger: logging.Logger, level: int = logging.INFO):
        self.logger = logger
        self.level = level
        self.buffer = ""  # 缓存未换行的分段内容

    def write(self, data: str) -> None:
        # 处理分块写入，按换行符拆分内容
        self.buffer += data
        lines = self.buffer.split('\n')
        # 保留最后一行（未完成的行）到缓存
        self.buffer = lines.pop() if lines else ""
        # 输出所有完整行
        for line in lines:
            if line:
                self.logger.log(self.level, line)

    def flush(self) -> None:
        # 输出缓存中剩余的未换行内容
        if self.buffer:
            self.logger.log(self.level, self.buffer)
            self.buffer = ""

2. 在invoke中使用包装器 #

import invoke
import logging

# 初始化日志配置（按需调整格式和级别）
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
log = logging.getLogger(__name__)

def print_foobar(context: invoke.context):
    # 创建LogStream实例，绑定目标日志器和日志级别
    log_stream = LogStream(log, logging.INFO)
    # 传入out_stream参数
    context.run("/usr/bin/printf 'foo\nbar\nbaz'", out_stream=log_stream)

关键说明 #

• write方法会缓存未换行的内容，确保只有完整的行才会被记录为日志，避免子进程分块输出导致的日志拆分。
• flush方法用于在程序结束或需要时，输出缓存中剩余的未完成行（比如没有换行符的结尾文本）。
• 可通过level参数灵活指定日志级别（如logging.DEBUG、logging.WARNING等）。

内容的提问来源于stack exchange，提问作者JamesThomasMoon