[Python] asyncio run coroutine from a synchronous function

Discussão em 'Python' iniciado por Stack, Junho 9, 2025.

  1. Stack

    Stack Membro Participativo

    How can I call task2 from func without declaring func async and awaiting it? My first thought was to create a thread and use run_coroutine_threadsafe but it deadlocks. Same as not using a thread. Do I have to start a new loop?

    import asyncio
    from threading import Thread

    async def task2():
    print("starting task2...")
    await asyncio.sleep(1)
    print("finished task2.")
    return "done"

    def func(loop=None):
    print("running func...")
    if not loop:
    loop = asyncio.get_running_loop()
    assert loop
    future = asyncio.run_coroutine_threadsafe(
    task2(),
    loop)
    result = future.result()
    print(f"{result=}")
    print("done func...")

    async def task1():
    print("starting task1...")
    await asyncio.sleep(1)
    # func()
    loop = asyncio.get_running_loop()
    t = Thread(target=func, args=(loop,))
    t.start()
    t.join()
    print("finished task1.")

    if __name__ == '__main__':
    asyncio.run(task1())

    Continue reading...

Compartilhe esta Página