stdio_proxy

stdio_proxy是一个线程安全的库，用于Python 2.7和Python 3.5+，可以将stdio临时重定向到另一个对象。

背景

Python 3.5+中的contextlib模块有redirect_stdout和redirect_stderr，分别是将sys.stdout和sys.stderr临时重定向到其他文件对象的有用函数。但是，这些函数对sys.stdout和sys.stderr有全局副作用。这意味着我们不能在大多数线程应用程序中使用这些函数。

• Python代码

import contextlib
import io
import time
from concurrent.futures import ThreadPoolExecutor

def run(value):
    for i in range(2):
        print("Hello from {}:{}".format(value, i))
        time.sleep(1)

def run_hook(value):
    buf = io.StringIO()
    with contextlib.redirect_stdout(buf):
        run(value)
    return buf.getvalue()

with ThreadPoolExecutor() as executor:
    f1 = executor.submit(run, "th1")
    f2 = executor.submit(run_hook, "th2")
    f1.result()
    result = f2.result()
    print("===Done===")
    print("Redirected Stdout:\n{}".format(result))

• 我们想要的功能

Hello from th1:0
Hello from th1:1
===Done===
Redirected Stdout:
Hello from th2:0
Hello from th2:1

• 实际输出的示例

Hello from th1:0
===Done===
Redirected Stdout:
Hello from th2:0
Hello from th1:1
Hello from th2:1

这个库旨在在线程应用程序中正确重定向这些stdio。只需用以下代码替换run_hook函数，结果将与“我们想要的功能”完全相同 :)。

def run_hook(value):
    buf = io.BytesIO()
    with stdio_proxy.redirect_stdout(buf):
        run(value)
    return buf.getvalue()

安装

$ pip install stdio-proxy

用法

• 将缓冲区重定向到stdin

buf = io.BytesIO(b"foo\n")
with stdio_proxy.redirect_stdin(buf):
    print("Read: {}".format(sys.stdin.readline()))

• 将stdout重定向到缓冲区

buf = io.BytesIO()
with stdio_proxy.redirect_stdout(buf):
    print("foo")
print("Redirected: {}".format(buf.getvalue()))

• 将stderr重定向到缓冲区

buf = io.BytesIO()
with stdio_proxy.redirect_stderr(buf):
    sys.stderr.write("foo\n")
print("Redirected: {}".format(buf.getvalue()))

许可

MIT许可