How to rerun page from worker thread?

[johnlyu]

My code looks like this

import time
from concurrent.futures import ThreadPoolExecutor

from queue import Queue

import streamlit as st



@st.cache_resource
def get_queue():
    return Queue()

queue = get_queue()


def worker(queue=queue):
    while True:
        queue.get()
        time.sleep(30)
        queue.task_done()
        st.rerun() # This is the key to make the app work

@st.cache_resource
def start_pool():
    pool = ThreadPoolExecutor(max_workers=1)
    pool.submit(worker, queue)

    return pool


start_pool()


submit = st.button("Submit")

if submit:
    queue.put(1)

st.write("Queue size: ", queue.qsize())

I have long run job in worker and want to auto refresh user’s page when the task is done.

But I got the warning 2024-10-22 15:35:14.247 Thread 'ThreadPoolExecutor-1_0': missing ScriptRunContext! This warning can be ignored when running in bare mode.

How can I send rerun signal to client browser from worker thread?

[adrien-turiot-maxa]

Here is a script with a worker that rerun the current session when some tasks are done:

import threading
import time

import streamlit as st
from streamlit.runtime import get_instance
from streamlit.runtime.scriptrunner import add_script_run_ctx
from streamlit.runtime.scriptrunner_utils.script_run_context import get_script_run_ctx


def main() -> None:
    if st.button("Start job"):
        start_background_worker()

    if "worker_results" not in st.session_state:
        st.session_state["worker_results"] = []

    for result in st.session_state["worker_results"]:
        st.write(result)


def start_background_worker() -> None:
    def start_work() -> None:
        ctx = get_script_run_ctx()
        assert ctx, "Context must be set with `add_script_run_ctx`."

        runtime = get_instance()

        task_todo = 1
        while task_todo < 5:
            print(f"Doing task {task_todo}...")
            time.sleep(1)

            ctx.session_state["worker_results"].append(f"Task {task_todo} done.")  # communicate through session state
            task_todo += 1

            if not runtime.is_active_session(ctx.session_id):
                print("The user quit the app.")
                runtime.close_session(ctx.session_id)
                return

        print("Rerun !")
        session_info = runtime._session_mgr.get_active_session_info(ctx.session_id)
        assert session_info, "Session must be active."
        session_info.session.request_rerun(None)

    thread = threading.Thread(target=start_work, daemon=True)
    add_script_run_ctx(thread)
    thread.start()


if __name__ == "__main__":
    main()