stdio-proxy

stdio_proxy

stdio_proxy is a thread-safe library for Python 2.7 and Python 3.5+ that can temporarily redirect stdio to another objects.

Background

Python 3.5+ has redirect_stdout and redirect_stderr in contextlib, that are utility functions for temporarily redirecting sys.stdout and sys.stderr to another file-like objects, respectively. But those functions have the global side effect on sys.stdout and sys.stderr. That means we cannot use those functions in most threaded applications.

• Python code

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))

• What we want

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

• Example of actual output

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

This library aims to redirect those stdio correctly in threaded applications as well. By just replacing run_hook function with the following code, the result would be exactly the same as "what we want" :)

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

Install

$ pip install stdio-proxy

Usage

• Redirect a buffer to stdin

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

• Redirect stdout to a buffer

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

• Redirect stderr to a buffer

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

License

MIT License