Running asyncio tasks in a separate thread under streamlit

async_tasks.py

import asyncio
import threading
from concurrent import futures
import streamlit as st


def create_async_loop():
    print("CREATE_ASYNC_LOOP START")
    loop = asyncio.new_event_loop()
    st.session_state.event_loop = loop
    done = asyncio.Event()
    st.session_state.event_loop_done = done

    async def loop_main():
        print("LOOP_MAIN START")
        await done.wait()
        print("LOOP_MAIN STOP")

    def thread_main():
        print("THREAD_MAIN START")
        loop.run_until_complete(loop_main())
        print("THREAD_MAIN STOP")

    loop_runner = threading.Thread(target=thread_main)
    st.session_state.event_loop_runner = loop_runner
    st.runtime.scriptrunner.add_script_run_ctx(
        loop_runner, st.runtime.scriptrunner.get_script_run_ctx()
    )
    loop_runner.start()
    print("CREATE_ASYNC_LOOP DONE")


def stop_async_loop():
    print("STOP_ASYNC_LOOP START")
    st.session_state.event_loop.call_soon_threadsafe(
        st.session_state.event_loop_done.set
    )
    st.session_state.event_loop_runner.join()
    del st.session_state.event_loop
    del st.session_state.event_loop_runner
    del st.session_state.event_loop_done
    print("STOP_ASYNC_LOOP DONE")


if "event_loop" not in st.session_state:
    create_async_loop()


async def do_nothing():
    print("DO_NOTHING START")
    await asyncio.sleep(3)
    print("DO_NOTHING DONE")


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

if st.button(label="Do something"):
    f = asyncio.run_coroutine_threadsafe(do_nothing(), st.session_state.event_loop)
    st.session_state.tasks.append(f)

if st.button(label="Do something else"):
    st.toast("Hello!")

if st.button(label="Stop"):
    stop_async_loop()
    for t in st.session_state.tasks:
        t.cancel()

taskbox = st.empty()
while st.session_state.tasks:
    st.session_state.tasks = [t for t in st.session_state.tasks if not t.done()]
    taskbox.write(st.session_state.tasks)
    futures.wait(st.session_state.tasks, timeout=1)
taskbox.write(st.session_state.tasks)