python - What is the pythonic way of running an asyncio event loop forever?

Question

From the docs it seems the recommended way to kickstart a asyncio application is to use asyncio.run(), so my application looks like this:

async def async_main():
    # Everything here can use asyncio.create_task():
    o = ObjectThatMustBeKeptReferenced()
    create_tasks_and_register_callbacks(o)

    # Wait forever, ugly:
    while True:
        await asyncio.sleep(10000)

asyncio.run(async_main())

This infinite loop at the end of async_main() feels very wrong. In other languages, there is where I would call the event loop forever. So I tried this:

def main():
    loop = asyncio.get_event_loop()

    # Everything here can use asyncio.create_task():
    o = ObjectThatMustBeKeptReferenced()
    create_tasks_and_register_callbacks(o)

    # Wait forever, pretty:
    loop.run_forever()

main()

The problem here is this will fail with an error of the sorts RuntimeError: no running event loop when I call asyncio.create_task() inside my functions, even though the event loop is created and registered on the thread.

What is the pythonic, one way of sleeping forever on the asyncio event loop?

Answer 1

You can simply change the sleep loop to an ad-hoc event which is never set:

# wait forever
await asyncio.Event().wait()

If needed, you can easily modify this to store the event into a variable and propagate it as a shutdown signal.

Another option is for your function that creates the tasks to return the tasks it has created, in which case you can await them even though (or precisely because) they'll never complete:

async def async_main():
    o = ObjectThatMustBeKeptReferenced()
    tasks = create_tasks_and_register_callbacks(o)
    # wait forever, or until a task raises
    await asyncio.gather(*tasks)

While this doesn't communicate the intention of looping forever as clearly, it has the advantage that it will halt the program (and propagate the exception) if any of the tasks raises an unhandled exception.