Как эффективно обрабатывать гнездовые асинхронные операции с помощью Python asyncio lib?

Мое первое наблюдение заключается в том, что вы не используете преимущества asyncio в момент написания вашего кода: вы создаете сопрограмму, немедленно планируете ее запуск и ожидаете ее завершения, прежде чем повторить это действие со следующей сопрограммой. То есть у вас не будет одновременно выполняться несколько задач asyncio, и обработка не будет перекрываться. Мы исправим это, создав три задачи, которые будут выполняться одновременно. Поскольку вы хотите, чтобы исключение, возникшее в одной задаче, привело к завершению других задач, я предлагаю использовать группу задач . Обратите внимание, что код можно упростить, если задача fetch_data_from_service вызывает process_data напрямую, и поэтому функция была переименована в fetch_and_process_data_from_service. Если process_data требует большого количества ресурсов ЦП, рассмотрите возможность выполнения этого кода в отдельном процессе с помощью run_in_executor.

import asyncio

async def fetch_and_process_data_from_service(service_name):
    await asyncio.sleep(1)  # Simulating I/O operation
    if service_name == "ServiceB":
        raise Exception(f"Error fetching data from {service_name}")
    return await process_data(f"Data from {service_name}")

async def process_data(data):
    await asyncio.sleep(1)  # Simulating data processing
    if data == "Data from ServiceC":
        raise Exception("Error processing data from ServiceC")
    return f"Processed {data}"

async def main_task():
    try:
        async with asyncio.TaskGroup() as tg:
            tasks = [
                tg.create_task(fetch_and_process_data_from_service(service_name))
                for service_name in ("ServiceA", "ServiceB", "ServiceC")
            ]

        results = [
            task.result() for task in tasks
        ]
        print(results)
    except Exception as e:
        print(f"Exception caught in main_task: {e}")
        raise  # propogate the exception back

# Running the event loop
if __name__ == "__main__":
    try:
        asyncio.run(main_task())
    except Exception as e:
        print(f"Exception caught in event loop: {e}")

Распечатки:

Exception caught in main_task: unhandled errors in a TaskGroup (1 sub-exception)
Exception caught in event loop: unhandled errors in a TaskGroup (1 sub-exception)

Однако учтите, что мы не знаем, какая задача не удалась и почему. Если это важно, мы можем запланировать запуск сопрограмм как параллельные задачи, используя `asyncio.gather', но становится необходимым явно отменить незавершенные задачи, если одна из задач вызывает исключение:

import asyncio

async def fetch_and_process_data_from_service(service_name):
    await asyncio.sleep(1)  # Simulating I/O operation
    if service_name == "ServiceB":
        raise Exception(f"Error fetching data from {service_name}")
    return await process_data(f"Data from {service_name}")

async def process_data(data):
    await asyncio.sleep(1)  # Simulating data processing
    if data == "Data from ServiceC":
        raise Exception("Error processing data from ServiceC")
    return f"Processed {data}"

async def main_task():
    tasks = [
        asyncio.create_task(fetch_and_process_data_from_service(service_name))
        for service_name in ("ServiceA", "ServiceB", "ServiceC")
    ]

    try:
        results = await asyncio.gather(*tasks)
        print(results)
    except Exception as e:
        print(f"Exception caught in main_task: {e}")
        for task in tasks:
            if not task.done():
                task.cancel()
        raise  # propogate the exception back

# Running the event loop
if __name__ == "__main__":
    try:
        asyncio.run(main_task())
    except Exception as e:
        print(f"Exception caught in event loop: {e}")

Распечатки:

Exception caught in main_task: Error fetching data from ServiceB
Exception caught in event loop: Error fetching data from ServiceB