装饰委托一个文件对象来添加功能

2022-01-18 00:00:00 python logging subprocess design-patterns

问题描述

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

I've been writing a small Python script that executes some shell commands using the subprocess module and a helper function:

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)

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

The following lines reads the standard output and error:

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

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

As you can see, stdout and stderr are not used. Suppose that I want to write the output and error messages to a log file, in a formatted way, e.g.:

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

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

My question is, what's the most Pythonic way to do it? I thought of three options:

  1. 继承文件对象并重写write方法.
  2. 使用实现 write 的 Delegate 类.
  3. 以某种方式连接到 PIPE 本身.
  1. Inherit the file object and override the write method.
  2. Use a Delegate class which implements write.
  3. Connect to the PIPE itself in some way.

更新:参考测试脚本

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

I'm checking the results with this script, saved as test.py:

#!/usr/bin/python
import sys

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

有什么想法吗?

解决方案

1 和 2 是合理的解决方案，但是重写 write() 是不够的.

1 and 2 are reasonable solutions, but overriding write() won't be enough.

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

The problem is that Popen needs file handles to attach to the process, so Python file objects doesn't work, they have to be OS level. To solve that you have to have a Python object that has a os level file handle. The only way I can think of solving that is to use pipes, so you have an os level file handle to write to. But then you need another thread that sits and polls that pipe for things to read in so it can log it. (So this is more strictly an implementation of 2, as it delegates to logging).

说到做到:

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'
' in buf:
                    data, buf = buf.split(b'
', 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 可以将标准输入/输出/错误附加到子进程.它还启动一个线程，该线程每秒轮询该管道的另一端以记录要记录的内容，然后使用日志记录模块进行记录.

So that class starts a os level pipe that Popen can attach the stdin/out/error to for the subprocess. It also starts a thread that polls the other end of that pipe once a second for things to log, which it then logs with the logging module.

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

Possibly this class should implement more things for completeness, but it works in this case anyway.

示例代码:

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

output.log 最终是这样的:

output.log ends up like so:

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

使用 Python 2.6、2.7 和 3.1 测试.

Tested with Python 2.6, 2.7 and 3.1.

我认为 1 和 3 的任何实现都需要使用类似的技术.这有点牵扯，但除非你能正确地自己制作 Popen 命令日志，否则我没有更好的主意).

I would think any implementation of 1 and 3 would need to use similar techniques. It is a bit involved, but unless you can make the Popen command log correctly itself, I don't have a better idea).

相关文章