Let's say, that I start a forever-running daemon in the background from main() using the task = asyncio.create_task(daemon_coro(signal)) where signal is an asyncio.Event(). The daemon_coro() sets the signal when there is an exception or when the daemon has successfully started.
Currently I'm detecting exceptions combining done() with exception():
tasks = set()
signal = asyncio.Event()
try:
task = asyncio.create_task(daemon_coro(signal))
tasks.add(task)
# Wait for the signal. daemon_coro() executes signal.set()
# in case of exceptions or in case the daemon was successfully
# started.
await signal.wait()
if task.done() and task.exception() is not None:
# There was an exception in daemon_coro(). Terminate the script.
raise asyncio.CancelledError
# If no exceptions were raised, then the daemon_coro()
# keeps running in the background and other coros
# are called here.
except asyncio.CancelledError:
pass
# Clean-up
for task in tasks:
task.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
Is it a robust solution and correct practice to use done() and exception() combined with asyncio.Event() to detect exceptions from forever-running tasks executed with asyncio.create_task()? I'm new to asyncio and thus I'm unsure.
You said you want to detect exceptions from from forever-running Tasks. Based on your code, however, I think a more precise description might be that you want to handle exceptions (or perhaps "react" to exceptions) from a forever-running Task.
I say this because "detecting" an exception has to happen in the Task itself. One of your comments says that you want to terminate the script when an exception occurs in the forever-running Task. That's what I mean by "handling" the exception.
The code you show doesn't really accomplish that. It only handles exceptions that occur right away, before the main Task reaches the line if task.done()... So you are only handling the case where the forever-Task is nice enough to fail quickly, before you check on it.
All the functionality you need is provided by asyncio.TaskGroup. This requires Python version >= 3.11. Here is a complete (but very simple) script. There is no need for a separate set to keep track of the Tasks to be canceled; that's done for you automatically. If the daemon Task fails, all the other Tasks in the group are canceled. You will even see a traceback so you can debug the exception that caused the problem.
import asyncio
async def daemon_coro(event):
event.set()
for _ in range(10):
await asyncio.sleep(0.25)
raise ValueError("Some exception")
async def main():
# If no exceptions were raised, then the daemon_coro()
# keeps running in the background and other coros
# are called here.
async with asyncio.TaskGroup() as group:
signal = asyncio.Event()
task = asyncio.create_task(daemon_coro(signal))
# Wait for the signal. daemon_coro() executes signal.set()
# when the daemon was successfully started.
await signal.wait()
for i in range(5):
task = group.create_task(asyncio.sleep(i+1))
tasks.add(task)
asyncio.run(main())
This script starts the daemon and waits for it to issue the signal, just like you did. Then it launches five more Tasks. After 2.5 seconds the daemon fails and issues an exception. The TaskGroup cancels all the other Tasks and propagates the exception into main(). I think that's just what you want.