如何在 Python 中异步记录 stdout/stderr?

Question

我有一个函数(我无法更改)，它将数据打印到stdout/stderr。我需要记录此函数的输出。我的第一个想法是用StringIO()形式的缓冲区替换sys.stdout，然后处理其内容。这工作得很好，但问题是，当函数失败时，它会打印错误消息并退出当前进程。在这种情况下，缓冲区的内容将丢失，因为我在函数调用后的代码永远不会执行。

所以我的想法是以某种方式异步监视缓冲区，并在有数据要读取时立即处理其内容。我尝试了使用asyncio及其add_reader方法的解决方案，但这似乎不支持StringIO()，甚至不支持常规文件。

这是我第一次尝试异步打印标准输出：

import asyncio
import sys
from io import StringIO
async def f():
print('Some output')
def logger(buffer):
sys.__stdout__.write(buffer.read())
buffer = StringIO()
sys.stdout = buffer
loop = asyncio.get_event_loop()
loop.add_reader(buffer, logger, buffer)
loop.run_until_complete(f())

失败了

ValueError: Invalid file object: <_io.StringIO object at 0x7f8a93e9aa68>

这个问题有什么解决方案吗？至少我需要澄清我的方法是否有意义。

更新：我发现了标准模块atexit，它可以在解释器退出时调用函数。这是解决我问题的另一种方法。

Answer 1

您可以创建io.TextIOBase的自定义子类，并将sys.stdout替换为自定义类的实例。 每当将输出发送到sys.stdout时，都会调用类的write()方法。 (可选)您可以将所有输出转发到原始标准输出：

class MyStdOut(io.TextIOBase):
def __init__(self, orig_stdout=None):
self.orig_stdout = orig_stdout
def write(self, s):
# Process output in whatever way you like
process(s)
# Write output to original stream, if desired
if self.orig_stdout:
self.orig_stdout.write(s)
sys.stdout = MyStdOut(sys.stdout)

这种方式将是完全同步的 - 不需要线程或异步I/O。