stdio_proxy is a thread-safe library for Python 2.7 and Python 3.5+ that can temporarily redirect stdio to another objects.
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
importcontextlibimportioimporttimefromconcurrent.futuresimportThreadPoolExecutordefrun(value):
foriinrange(2):
print("Hello from {}:{}".format(value, i))
time.sleep(1)
defrun_hook(value):
buf=io.StringIO()
withcontextlib.redirect_stdout(buf):
run(value)
returnbuf.getvalue()
withThreadPoolExecutor() asexecutor:
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:1This 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" :)
defrun_hook(value):
buf=io.BytesIO()
withstdio_proxy.redirect_stdout(buf):
run(value)
returnbuf.getvalue()$ pip install stdio-proxy- Redirect a buffer to
stdin
buf=io.BytesIO(b"foo\n")
withstdio_proxy.redirect_stdin(buf):
print("Read: {}".format(sys.stdin.readline()))- Redirect
stdoutto a buffer
buf=io.BytesIO()
withstdio_proxy.redirect_stdout(buf):
print("foo")
print("Redirected: {}".format(buf.getvalue()))- Redirect
stderrto a buffer
buf=io.BytesIO()
withstdio_proxy.redirect_stderr(buf):
sys.stderr.write("foo\n")
print("Redirected: {}".format(buf.getvalue()))MIT License