如何让Python的asyncio强制记录任务异常，不受await与垃圾回收影响？

I totally get where you're coming from—coming from Dart's async model, where unhandled async exceptions are hard to miss, Python's default behavior here feels like a total violation of the Zen of Python's 10th rule ("Errors should never pass silently"). Let's break down how to fix this so you can safely run "fire-and-forget" background tasks without worrying about hidden exceptions.

The Core Problem #

As you noted, when you call Future.set_exception() or a task throws an exception but is never awaited, asyncio won't propagate that exception to your user code by default. It only logs it when the Future/Task is garbage-collected, which could be long after the error occurred, or never if the object is still referenced somewhere.

Solutions to Force Immediate Exception Logging #

1. Use a "Fire-and-Forget" Wrapper with Exception Catching #

The cleanest approach is to wrap your coroutines in a helper function that automatically catches and logs exceptions before submitting them to the event loop. This way, you don't have to remember to add callbacks or handle exceptions manually every time.

import asyncio
import logging

# Configure logging first (adjust level/format as needed)
logging.basicConfig(
    level=logging.ERROR,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)

def fire_and_forget(coro):
    async def wrapped_coroutine():
        try:
            await coro
        except Exception as e:
            # Log the full exception stack trace for debugging
            logging.error(
                "Unhandled exception in fire-and-forget task",
                exc_info=True
            )
    # Submit the wrapped coroutine to the event loop
    asyncio.create_task(wrapped_coroutine())

# Example usage
async def problematic_task():
    raise ValueError("Something went wrong in the background!")

async def main():
    # Just call the wrapper—no need to await
    fire_and_forget(problematic_task())
    await asyncio.sleep(1)  # Give the task time to run

asyncio.run(main())

This wrapper ensures that any exception thrown in your background task is caught immediately and logged with a full stack trace, no matter if you ever await the task or not.

2. Add a Done Callback to Tasks #

If you prefer more control, you can attach a callback to any asyncio.Task that checks for exceptions and logs them as soon as the task completes.

import asyncio
import logging

logging.basicConfig(level=logging.ERROR)

def log_task_exception(task):
    exc = task.exception()
    if exc is not None:
        logging.error(
            "Task failed with exception",
            exc_info=exc
        )

# Example usage
async def main():
    task = asyncio.create_task(problematic_task())
    task.add_done_callback(log_task_exception)
    await asyncio.sleep(1)

asyncio.run(main())

This works well if you need to handle exceptions differently for specific tasks, but it's a bit more verbose than the wrapper approach for "fire-and-forget" use cases.

3. Customize the Event Loop's Exception Handler #

You can set a global exception handler for the asyncio event loop to catch unhandled exceptions from tasks that don't have explicit callbacks or awaiters. Note that by default, asyncio will still delay logging until the task is garbage-collected, but you can work around this by combining it with task tracking.

import asyncio
import logging

logging.basicConfig(level=logging.ERROR)

def global_exception_handler(loop, context):
    # Log the exception details
    logging.error(
        "Unhandled asyncio exception",
        exc_info=context.get("exception")
    )
    # Let the default handler run too (optional)
    loop.default_exception_handler(context)

async def main():
    loop = asyncio.get_running_loop()
    loop.set_exception_handler(global_exception_handler)
    
    asyncio.create_task(problematic_task())
    await asyncio.sleep(1)

asyncio.run(main())

While this covers global cases, it's still dependent on the task being garbage-collected to trigger the handler. For true immediate logging, the wrapper or done callback approaches are more reliable.

Why This Differs From Dart #

Dart's runtime treats unhandled async exceptions as critical errors that will crash the app (by default), which aligns with the "no silent errors" principle. Python's asyncio was designed with more flexibility, but that flexibility can lead to hidden bugs if you're not careful. The solutions above bring Python's behavior closer to what you're used to in Dart.

内容的提问来源于stack exchange，提问作者Dev Aggarwal