python - 装饰\委托(delegate)一个文件对象来添加功能

Question

我一直在编写一个小的 Python 脚本，它使用 subprocess 模块和一个辅助函数来执行一些 shell 命令:

import subprocess as sp
def run(command, description):
    """Runs a command in a formatted manner. Returns its return code."""
    start=datetime.datetime.now()
    sys.stderr.write('%-65s' % description)
    s=sp.Popen(command, shell=True, stderr=sp.PIPE, stdout=sp.PIPE)
    out,err=s.communicate()
    end=datetime.datetime.now()
    duration=end-start
    status='Done' if s.returncode==0 else 'Failed'
    print '%s (%d seconds)' % (status, duration.seconds)

以下行读取标准输出和错误:

    s=sp.Popen(command, shell=True, stderr=sp.PIPE, stdout=sp.PIPE)
    out,err=s.communicate()

如您所见，未使用 stdout 和 stderr。假设我想以格式化的方式将输出和错误消息写入日志文件，例如:

[STDOUT: 2011-01-17 14:53:55] <message>
[STDERR: 2011-01-17 14:53:56] <message>

我的问题是，最符合 Pythonic 的方法是什么？我想到了三个选项:

1. 继承文件对象并覆盖write方法。
2. 使用实现 write 的委托(delegate)类。
3. 以某种方式连接到 PIPE 本身。

更新:引用测试脚本

我正在用这个脚本检查结果，保存为 test.py:

#!/usr/bin/python
import sys

sys.stdout.write('OUT\n')
sys.stdout.flush()
sys.stderr.write('ERR\n')
sys.stderr.flush()

有什么想法吗？

Answer 1

1 和 2 是合理的解决方案，但覆盖 write() 是不够的。

问题是 Popen 需要文件句柄来附加到进程，所以 Python 文件对象不起作用，它们必须是操作系统级别的。要解决这个问题，您必须拥有一个具有操作系统级别文件句柄的 Python 对象。我能想到的解决这个问题的唯一方法是使用管道，这样你就有了一个操作系统级别的文件句柄来写入。但是随后您需要另一个线程来轮询该管道以获取要读入的内容，以便它可以记录它。 (所以这更严格地是 2 的实现，因为它委托(delegate)给日志记录)。

说完了:

import io
import logging
import os
import select
import subprocess
import time
import threading

LOG_FILENAME = 'output.log'
logging.basicConfig(filename=LOG_FILENAME,level=logging.DEBUG)

class StreamLogger(io.IOBase):
    def __init__(self, level):
        self.level = level
        self.pipe = os.pipe()
        self.thread = threading.Thread(target=self._flusher)
        self.thread.start()

    def _flusher(self):
        self._run = True
        buf = b''
        while self._run:
            for fh in select.select([self.pipe[0]], [], [], 0)[0]:
                buf += os.read(fh, 1024)
                while b'\n' in buf:
                    data, buf = buf.split(b'\n', 1)
                    self.write(data.decode())
            time.sleep(1)
        self._run = None

    def write(self, data):
        return logging.log(self.level, data)

    def fileno(self):
        return self.pipe[1]

    def close(self):
        if self._run:
            self._run = False
            while self._run is not None:
                time.sleep(1)
            os.close(self.pipe[0])
            os.close(self.pipe[1])

因此该类启动了一个操作系统级别的管道，Popen 可以将标准输入/输出/错误附加到子进程。它还会启动一个线程，该线程每秒轮询一次该管道的另一端以记录要记录的内容，然后使用日志记录模块进行记录。

为了完整性，这个类可能应该实现更多的东西，但无论如何它都适用于这种情况。

示例代码:

with StreamLogger(logging.INFO) as out:
    with StreamLogger(logging.ERROR) as err:
        subprocess.Popen("ls", stdout=out, stderr=err, shell=True)

output.log 最终是这样的:

INFO:root:output.log
INFO:root:streamlogger.py
INFO:root:and
INFO:root:so
INFO:root:on

已使用 Python 2.6、2.7 和 3.1 进行测试。

我认为 1 和 3 的任何实现都需要使用类似的技术。这有点复杂，但除非您可以使 Popen 命令本身正确记录，否则我没有更好的主意)。