How can exceptions be caught when working with…

Using try...except blocks.

import asyncio

async def my_async_function():
    # Some asynchronous code that may raise an exception
    await asyncio.sleep(1)
    raise ValueError("Something went wrong!")

async def main():
    try:
        await my_async_function()
    except ValueError as e:
        print(f"Caught exception: {e}")
    except Exception as e:
        print(f"Caught unexpected exception: {e}")

if __name__ == "__main__":
    asyncio.run(main())

In the case of running multiple asynchronous tasks concurrently with asyncio.gather, exceptions can be handled as follows:

import asyncio

async def task_with_exception():
    await asyncio.sleep(0.5)
    raise RuntimeError("Error in one of the tasks")

async def task_without_exception():
    await asyncio.sleep(1)
    print("Task completed without exception")

async def concurrent_execution():
    try:
        # return_exceptions=True allows collecting all exceptions instead of stopping at the first
        results = await asyncio.gather(
            task_with_exception(),
            task_without_exception(),
            return_exceptions=True
        )
        for result in results:
            if isinstance(result, Exception):
                print(f"Exception detected in task: {result}")
            else:
                print(f"Task completed successfully with result: {result}")

    except Exception as e:
        # This block executes only if return_exceptions=False and an exception occurs before handling
        print(f"Exception during gather execution: {e}")

if __name__ == "__main__":
    asyncio.run(concurrent_execution())

The return_exceptions=True parameter in asyncio.gather allows collecting exceptions as results instead of stopping all tasks at the first exception.

When using asyncio.create_task or similar low-level mechanisms, exceptions can be retrieved using the .exception() method of the task object after it completes, if they were not handled inside the coroutine:

import asyncio

async def failing_task():
    await asyncio.sleep(0.1)
    raise TypeError("Incorrect type!")

async def monitor_task():
    task = asyncio.create_task(failing_task())
    await task # Wait for task to complete
    if task.done():
        exception = task.exception()
        if exception:
            print(f"Task finished with exception: {exception}")
        else:
            print("Task finished without exception")

if __name__ == "__main__":
    asyncio.run(monitor_task())

How can exceptions be caught when working with asynchronous code?