Capturing stdout, stderr from a running thread at runtime

Question

I'm trying to capture stdout and stderr from an exec() call at runtime i.e. before it completes. I have wrapped it in a thread under a contextlib context to redirect output but with no success. I'm trying to avoid subprocessing. The following code depicts a simplified version of what I'm trying to do.

import threading
from io import StringIO
import contextlib

code = "import time; print('hello'); time.sleep(2); print('working'); time.sleep(2); print('coming to an end'); time.sleep(2); print('end')"
f = StringIO()
e = StringIO()
    
with contextlib.redirect_stdout(f), contextlib.redirect_stderr(e):
    t1 = threading.Thread(target=exec, args=(code, globals(), locals()))
    t1.start()
    while t1.is_alive(): 
        print(f.getvalue())
        print(e.getvalue())
    t1.join()

Answer 1

If you use print with the default file parameter, then it will print to sys.stdout which we have changed with contextlib.redirect_stdout to f causing your code to write the result of f.getvalue() back to f. So the following statements are almost equivalant within the context.

>>> print(f.getvalue())
>>> f.write(f.getvalue())

You can reach out to the normal stdout by using file=sys.__stdout__. that should not be overridden with contextlib.redirect_stdout. If you want to see output immediately, then you might want to set flush to True. Further f.getvalue() should have its own newlines so our print does not have to append newline so we set end="".

>>> print(f.getvalue(), file=sys.__stdout__, flush=True, end="")

But now you might get a new problem: While the thread is active you print the entire StringIO buffer as often as python can: "hello, hello, ..., hello working, hello working ...". You can solve this in two ways. Either you wait until the thread is completed and you print f.getvalue() once. In that case you dont even have to bother with sys.__stdout__. Because you can just put your print(f.getvalue()) outside the stdout context. But this means you will not have access to the text until the thread is completed. Depending on your use case that can be anoying because the text is already writen to the buffer. So another way you can solve this is to overwrite the write method of the file-like object that you replace stdout with. that could look something like this:

from contextlib import redirect_stdout
from io import StringIO
import sys
import threading

class get_and_print(StringIO):
    def write(self, *args):
        print(*args, file=sys.__stdout__, flush=True, end="") # print directly to the terminal
        return super().write(*args) # update the internal buffer using the write method of the base class StringIO

code = "import time; print('hello'); time.sleep(2); print('working'); time.sleep(2); print('coming to an end'); time.sleep(2); print('end')"

f = get_and_print()

with redirect_stdout(f):
    t1 = threading.Thread(target=exec, args=(code, globals(), locals()))
    t1.start()
    # we just wait until the tread completes because t1 is responsible for printing to sys.stdout
    t1.join()

if False:
    # we still can print f.getvalue later
    print(f.getvalue())

Now you will have access to the the new text as soon as it is written. But you can still call f.getvalue later to get all the text, once the thread is finished.