Как можно отловить исключения при работе с асинхронным кодом?

Используя try...except блоки.

import asyncio

async def my_async_function():
    # Какой-то асинхронный код, который может вызвать исключение
    await asyncio.sleep(1)
    raise ValueError("Что-то пошло не так!")

async def main():
    try:
        await my_async_function()
    except ValueError as e:
        print(f"Поймано исключение: {e}")
    except Exception as e:
        print(f"Поймано непредвиденное исключение: {e}")

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

В случае конкурентного запуска нескольких асинхронных задач с использованием asyncio.gather, исключения могут обрабатываться следующим образом:

import asyncio

async def task_with_exception():
    await asyncio.sleep(0.5)
    raise RuntimeError("Ошибка в одной из задач")

async def task_without_exception():
    await asyncio.sleep(1)
    print("Задача без исключения завершена")

async def concurrent_execution():
    try:
        # return_exceptions=True позволяет собрать все исключения вместо падения при первом же
        results = await asyncio.gather(
            task_with_exception(),
            task_without_exception(),
            return_exceptions=True
        )
        for result in results:
            if isinstance(result, Exception):
                print(f"Обнаружено исключение в задаче: {result}")
            else:
                print(f"Задача успешно завершена с результатом: {result}")

    except Exception as e:
        # Этот блок выполнится, только если return_exceptions=False и 
        # одна из задач выбросила исключение до того, как мы его поймали внутри gather
        print(f"Обнаружено исключение при выполнении gather: {e}")

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

Параметр return_exceptions=True в asyncio.gather позволяет собрать исключения как результаты выполнения, вместо того, чтобы остановить выполнение всех задач при возникновении первого исключения.

При использовании asyncio.create_task или подобных низкоуровневых механизмов, исключения могут быть получены с помощью метода .exception() у объекта задачи после её завершения, если они не были обработаны внутри самой корутины:

import asyncio

async def failing_task():
    await asyncio.sleep(0.1)
    raise TypeError("Неправильный тип!")

async def monitor_task():
    task = asyncio.create_task(failing_task())
    await task # Ждем завершения задачи
    if task.done():
        exception = task.exception()
        if exception:
            print(f"Задача завершилась с исключением: {exception}")
        else:
            print("Задача завершилась без исключения")

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