В чем разница между TaskGroup и gather в asyncio?

TaskGroup vs gather в asyncio

Оба метода используются для одновременного запуска асинхронных задач, но они кардинально различаются подходом к обработке ошибок и управлению жизненным циклом.

gather() — старый способ

Одновременный запуск корутин с контролем ошибок через параметры:

import asyncio

async def fetch_data(url):
    await asyncio.sleep(1)
    return f"Data from {url}"

async def main_gather():
    # Запускаем несколько корутин параллельно
    results = await asyncio.gather(
        fetch_data("url1"),
        fetch_data("url2"),
        fetch_data("url3")
    )
    print(results)  # ["Data from url1", "Data from url2", "Data from url3"]

asyncio.run(main_gather())

TaskGroup() — современный способ (Python 3.11+)

Текстовый менеджер для управления группой задач с встроенной поддержкой исключений:

import asyncio

async def main_taskgroup():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(fetch_data("url1"))
        task2 = tg.create_task(fetch_data("url2"))
        task3 = tg.create_task(fetch_data("url3"))
    
    # После выхода из контекста все задачи завершены
    print(task1.result(), task2.result(), task3.result())

asyncio.run(main_taskgroup())

Основные различия

1. Обработка исключений

Eсли одна из задач выбросит исключение:

async def failing_task():
    await asyncio.sleep(0.5)
    raise ValueError("Oops!")

async def success_task():
    await asyncio.sleep(1)
    return "Success"

# gather() с return_exceptions=False (по умолчанию)
async def gather_example():
    try:
        results = await asyncio.gather(
            failing_task(),
            success_task(),
            return_exceptions=False
        )
    except ValueError as e:
        print(f"Поймали ошибку: {e}")
        # success_task всё ещё выполняется в фоне!
        await asyncio.sleep(2)  # Ждём завершения

# TaskGroup() — поддерживает ExceptionGroup
async def taskgroup_example():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(failing_task())
            tg.create_task(success_task())
    except ExceptionGroup as eg:
        print(f"Группа ошибок: {eg.exceptions}")
        # Обе задачи гарантированно завершены

2. Явное управление задачами

# gather() — передаёшь корутины напрямую
results = await asyncio.gather(
    task1_coro(),
    task2_coro(),
    task3_coro()
)

# TaskGroup() — создаёшь задачи явно
async with asyncio.TaskGroup() as tg:
    task1 = tg.create_task(task1_coro())
    task2 = tg.create_task(task2_coro())
    task3 = tg.create_task(task3_coro())

3. Отмена задач

# gather() — нужно отслеживать Task
tasks = [asyncio.create_task(fetch(url)) for url in urls]
results = await asyncio.gather(*tasks)

# При ошибке нужно вручную отменять
for task in tasks:
    if not task.done():
        task.cancel()

# TaskGroup() — автоматически отменяет при ошибке
async with asyncio.TaskGroup() as tg:
    for url in urls:
        tg.create_task(fetch(url))
    # При исключении все задачи отменяются автоматически

Практический пример: обработка множества запросов

import asyncio
import aiohttp

async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.json()

async def gather_approach(urls):
    async with aiohttp.ClientSession() as session:
        try:
            results = await asyncio.gather(
                *[fetch_url(session, url) for url in urls],
                return_exceptions=True  # Ловим ошибки
            )
            return [r for r in results if not isinstance(r, Exception)]
        except Exception as e:
            print(f"Ошибка: {e}")
            # Некоторые запросы могут остаться в памяти

async def taskgroup_approach(urls):
    async with aiohttp.ClientSession() as session:
        try:
            results = []
            async with asyncio.TaskGroup() as tg:
                for url in urls:
                    task = tg.create_task(fetch_url(session, url))
                    results.append(task)
            return [task.result() for task in results]
        except ExceptionGroup as eg:
            print(f"Ошибки: {eg.exceptions}")
            # Все задачи завершены
            return []

Когда использовать?

gather():

• Поддержка Python < 3.11
• Простые случаи без сложной обработки ошибок
• Когда нужно вернуться к выполнению после ошибки (return_exceptions=True)

TaskGroup():

• Python 3.11+ (рекомендуется)
• Критична корректная обработка ошибок
• Нужно гарантировать завершение всех задач
• Хочешь явно управлять жизненным циклом
• Нужна отмена задач при ошибках

Итоговое сравнение

TaskGroup — это современный стандарт, который решает боль-точки gather(). Если работаешь на Python 3.11+, используй TaskGroup.