我有一个函数(我无法更改),它将数据打印到 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
,它可以在解释器退出时调用函数。这是解决我的问题的另一种方法。
最佳答案
您可以创建 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。
关于python - 如何在Python中异步记录stdout/stderr?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/49071558/