app.py

#!/usr/bin/env python
"""Observe the stdout redirected from sub thread to main thread through shared StringIO instance"""

import time
import contextlib

from concurrent.futures import ThreadPoolExecutor
from io import StringIO

import streamlit as st


def print_func(sec: int):
    for i in range(sec):
        print(f"{i} [sec] printed in sub thread\n")
        time.sleep(1)
    return "print finished"

def thread_func(sio: StringIO):
    with contextlib.redirect_stdout(sio):
        result = print_func(10)
        return result


if __name__ == '__main__':
    st.write("# thread test")

    shared_sio = StringIO()

    with ThreadPoolExecutor() as executor:
        future = executor.submit(thread_func, shared_sio)

        # observe shared_sio
        placeholder = st.empty()
        while future.running():
            placeholder.empty()
            placeholder.write(shared_sio.getvalue())
            time.sleep(1)

        placeholder.empty()
        placeholder.write(shared_sio.getvalue())

        result = future.result()

        st.write(result)