Python: Handling exceptions when using async/await

This concise, straight-to-the-point article will walk you through some different ways to deal with exceptions when using async/await in Python

Using try/except blocks

You can use the standard try/except blocks to catch and handle exceptions that may occur in async functions or await expressions. The main points are:

1. Place the asynchronous code that may raise an exception within a try block.
2. In the except block, handle the specific exception(s) that you expect to occur.
3. Optionally, use a general except block to handle any unexpected exceptions.
4. Handle the exception appropriately, such as logging an error, retrying the operation, or raising a different exception.

Example:

import asyncio

# Define an async function
async def async_function():
    try:
        # Async code that may raise an exception
        await asyncio.sleep(1)
        result = await async_operation()
    except asyncio.TimeoutError:
        print("Timeout error occurred.")
    except Exception as e:
        print(f"An unexpected error occurred: {str(e)}")

# This async function will raise an exception
async def async_operation():
    await asyncio.sleep(2)
    raise ValueError("Something went wrong.")

# Main function
async def main():
    try:
        await async_function()
    except Exception as e: 
        print(f"Error in main function: {str(e)}")

# Run the main function
asyncio.run(main())

Output:

An unexpected error occurred: Something went wrong.

Some thoughts:

• Pros: This approach is simple, familiar, and explicit. It allows fine-grained control over which exceptions to catch and how to handle them.
• Cons: This approach may be verbose and repetitive if there are many async functions or await expressions that may raise exceptions. It may also obscure the main logic of the code with too many try/except blocks.

Using asyncio.gather()

This approach involves using asyncio.gather() to handle exceptions when using async/await in Python. It enables concurrent execution of multiple async tasks and provides a simpler way to collect their results and handle exceptions.

1. Create a list of coroutines or futures representing the async tasks you want to execute.
2. Use asyncio.gather() to concurrently execute the tasks.
3. Await the result of asyncio.gather() to retrieve the results and handle exceptions, if any.

Code example:

import asyncio

async def async_task1():
    await asyncio.sleep(2)
    raise ValueError("Error in async_task1")

async def async_task2():
    await asyncio.sleep(3)
    return "Result from async_task2"

async def main():
    tasks = [async_task1(), async_task2()]
    try:
        results = await asyncio.gather(*tasks)
        print(f"Results: {results}")
    except ValueError as e:
        print(f"A value error occurred: {str(e)}")
    except Exception as e:
        print(f"Something went wrong: {str(e)}")

asyncio.run(main())

Output:

A value error occurred: Error in async_task1

Pros & cons:

• Pros: Enables concurrent execution of multiple async tasks and simplifies exception handling by automatically raising the first exception encountered.
• Cons: Only raises the first exception, potentially hiding subsequent exceptions.

That’s it. Happy coding & enjoy your day!