Skip to main content

Python Async Error Handling: Exceptions in asyncio

Asynchronous programming introduces new complexities in error handling. When you run dozens of concurrent tasks, a single failure can crash your application if not handled properly. This guide covers essential patterns for handling exceptions in asyncio to build robust and resilient concurrent applications that gracefully handle failures without losing data or leaving orphaned tasks.

Basic Error Handling in a Coroutine​

At its core, handling an error inside a single coroutine works exactly like it does in synchronous code: you use a try...except block. The exception handling logic remains the same; the only difference is that the code being protected is asynchronous.

import asyncio

async def might_fail(should_succeed: bool):
    """A coroutine that simulates an operation that might fail."""
    print(f"Executing operation that should {'succeed' if should_succeed else 'fail'}...")
    await asyncio.sleep(1)
    if not should_succeed:
        raise ValueError("The operation failed!")
    return "Success!"

async def main():
    try:
        # Await the coroutine inside a try block
        result = await might_fail(should_succeed=False)
        print(f"Result: {result}")
    except ValueError as e:
        print(f"Caught a specific error: {e}")

asyncio.run(main())

Output:

Executing operation that should fail...
Caught a specific error: The operation failed!

This pattern is straightforward: the try...except block catches the ValueError raised from the awaited coroutine, and the program continues executing. The await keyword ensures that the exception propagates to the exception handler immediately when it's raised.

Handling Errors in Concurrent Tasks with asyncio.gather()​

The real challenge in async error handling arises when you run multiple tasks concurrently with asyncio.gather(). Understanding how gather() handles exceptions is critical for writing resilient concurrent applications.

The Default Behavior: Fail Fast​

By default, if any task submitted to gather() raises an exception, gather() immediately propagates that exception without waiting for other tasks to complete. This "fail-fast" behavior can leave other tasks running in the background.

# This will crash the program as soon as the ValueError is raised.
# The other tasks will continue running until the program exits.
# results = await asyncio.gather(
#     might_fail(True),
#     might_fail(False),  # This will raise an exception
#     might_fail(True)
# )

The Robust Solution: return_exceptions=True​

To handle this gracefully, asyncio.gather() provides the return_exceptions parameter. When set to True, gather() will not raise exceptions immediately. Instead, it treats exceptions as successful results and includes the exception objects themselves in the returned list. This allows you to wait for all tasks to complete and then handle successes and failures individually.

import asyncio

async def might_fail(should_succeed: bool):
    await asyncio.sleep(1)
    if not should_succeed:
        raise ValueError("This operation failed!")
    return "Success!"

async def main_robust():
    print("Running multiple tasks concurrently...")

    # return_exceptions=True tells gather to not raise errors immediately
    results = await asyncio.gather(
        might_fail(True),
        might_fail(False),
        might_fail(True),
        return_exceptions=True
    )

    print("\n--- All tasks have completed ---")
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Task {i} failed with an error: {result}")
        else:
            print(f"Task {i} succeeded with result: {result}")

asyncio.run(main_robust())

Output:

Running multiple tasks concurrently...

--- All tasks have completed ---
Task 0 succeeded with result: Success!
Task 1 failed with an error: This operation failed!
Task 2 succeeded with result: Success!

This pattern is essential for building resilient applications. You can process all successful results while logging, retrying, or alerting for failures. According to the asyncio documentation, return_exceptions=True is the recommended approach when you want to ensure all tasks complete regardless of individual failures.

Why this approach is superior:

  • Completion Guarantee: All tasks run to completion, giving you a complete picture of what succeeded and what failed.
  • Selective Handling: You decide what to do with each failure independently—retry, log, alert, or aggregate them.
  • Production Ready: This pattern matches the error handling style of real-world concurrent systems that need to tolerate partial failures.

The "Task Exception Was Never Retrieved" Warning​

A common pitfall for newcomers to asyncio is the "fire-and-forget" anti-pattern. This occurs when you create a task but never await it or check its result. If that task raises an exception, the warning "Task exception was never retrieved" appears when the program exits.

async def fire_and_forget():
    # This creates a task, but we never store a reference or await it
    asyncio.create_task(might_fail(False))
    # The main coroutine finishes before the task does
    await asyncio.sleep(2)

# When this program exits, you'll see a warning like:
# "Task exception was never retrieved"
# asyncio.run(fire_and_forget())

This warning signals a potential bug: an error occurred silently without any handling, recovery, or notification. It's a sign that your error handling logic is incomplete.

How to Fix It: Three Approaches​

1. Await the task directly (most common):

async def proper_error_handling():
    try:
        result = await asyncio.create_task(might_fail(False))
    except ValueError as e:
        print(f"Handled error: {e}")

2. Use asyncio.gather() (recommended for multiple tasks):

async def gather_approach():
    results = await asyncio.gather(
        might_fail(True),
        might_fail(False),
        return_exceptions=True
    )

3. Add a done callback (for true background tasks):

async def background_task_with_callback():
    task = asyncio.create_task(might_fail(False))
    
    def handle_done(t):
        try:
            t.result()  # Re-raise the exception if one occurred
        except Exception as e:
            print(f"Background task failed: {e}")
    
    task.add_done_callback(handle_done)
    # Now the task runs in the background, but errors are logged

The done callback approach is useful for true background tasks where you don't want to block waiting for completion, but still want to log errors.

Key Takeaways​

  • Simple Pattern: Use try...except blocks within coroutines for basic exception handling, just as you would in synchronous code.
  • Concurrent Safety: Use asyncio.gather(..., return_exceptions=True) to run multiple tasks safely and handle failures without crashing the entire group.
  • Always Handle Tasks: Never create a task with create_task() without a plan to await it or attach a callback. Unhandled exceptions are silent bugs.
  • Choose Your Pattern: Use direct await for single tasks, gather() for multiple tasks, and callbacks for true background work.
  • Production Resilience: Proper async error handling prevents silent failures and enables graceful degradation when individual tasks fail.

Frequently Asked Questions​

What is the difference between await and create_task()?​

await coroutine() directly executes a coroutine and waits for completion before continuing. asyncio.create_task(coroutine()) schedules a coroutine to run concurrently and immediately returns a Task object, allowing other code to run. Use await for sequential execution; use create_task() to start background work. Always store the task or ensure its result is checked later.

Can I have multiple except blocks in a coroutine?​

Yes, coroutines support multiple exception handlers exactly like regular functions. Catch specific exceptions first, then more general ones: except ValueError before except Exception. The first matching handler executes. This allows fine-grained error handling and different recovery strategies for different failure types.

How do I retry a failed async operation?​

Wrap the coroutine in a loop that re-executes on failure, with optional backoff and a maximum retry count. For example: a for-loop that awaits the operation, breaks on success, and continues on Exception. For production code, use a library like tenacity that provides decorators for retries with exponential backoff and jitter.

What happens if gather() itself raises an exception?​

gather() itself can raise exceptions related to the gather operation itself, not from the tasks. For example, invalid arguments cause TypeError. The more common scenario is exceptions raised by the tasks, which are either propagated immediately (default) or returned in the results list (return_exceptions=True). Always wrap gather() in try/except for both scenarios.

Further Reading​

Next Steps​

You now have the tools to write concurrent code that is both fast and resilient to failures. Next, we'll explore how to interact with external resources in an asynchronous world, starting with "Working with Async HTTP Requests: aiohttp."

Happy (safe) coding!