TaskGroup в Python 3.11: почему он вытесняет gather

Вы запускаете десять корутин через asyncio.gather, третья падает с исключением, а остальные девять продолжают лезть в базу и жечь соединения. Через минуту вы находите в логах Task was destroyed but it is pending и понимаете, что часть запросов так и не завершилась. Это не баг в вашем коде — это поведение gather по умолчанию. С Python 3.11 появился asyncio.TaskGroup, который на подобной ошибке отменяет все соседние задачи и не даёт им утечь в фон.

Разберём, чем именно TaskGroup отличается от gather, где старый подход даёт незаметные проблемы и когда переходить не нужно.

Что ломается в gather

У asyncio.gather есть параметр return_exceptions, и оба его значения ведут себя неудобно, если вам нужна надёжность.

При return_exceptions=False (по умолчанию) первое же исключение пробрасывается наверх сразу. Но остальные задачи не отменяются — они остаются запущенными в цикле событий. Вы получили ошибку, обработали её, а фоновые корутины продолжают работать и могут упасть уже без обработчика.

При return_exceptions=True gather дожидается всех задач и складывает исключения в список результатов вместе с нормальными значениями. Ничего не отменяется, ничего не пробрасывается — вы обязаны сами перебрать результаты и найти те, что оказались экземплярами Exception. Забыли проверить — ошибка проглочена молча.

gather решает задачу «дождись всех корутин». Он не решает задачу «если одна упала, аккуратно останови остальные» — а именно это нужно почти всегда в реальном коде.

Второй тонкий момент — отмена. Если внешний код отменяет корутину, внутри которой висит gather, отмена доходит до дочерних задач не всегда предсказуемо: поведение с распространением CancelledError в gather исторически было источником путаницы между версиями. Проверяйте на своей версии, но общее правило — gather не даёт гарантии, что после выхода из него в фоне ничего не осталось.

Как выглядит проблема в коде

# Python 3.11+
import asyncio

async def fetch(name, delay, fail=False):
    await asyncio.sleep(delay)
    if fail:
        raise ValueError(f"{name} упал")
    print(f"{name} завершился")
    return name

async def main_gather():
    try:
        await asyncio.gather(
            fetch("a", 0.1, fail=True),   # упадёт первой
            fetch("b", 0.5),              # ещё работает
            fetch("c", 0.5),              # ещё работает
        )
    except ValueError as exc:
        print(f"Поймали: {exc}")
    # b и c к этому моменту НЕ отменены и продолжат печатать
    await asyncio.sleep(1)

asyncio.run(main_gather())

Вывод покажет строку «Поймали: a упал», а затем — «b завершился» и «c завершился». Задачи, которые вам уже не нужны, доработали до конца. На запросах в сеть или в базу это лишняя нагрузка и потенциальные записи, которых вы не ждали.

Как это делает TaskGroup

TaskGroup — асинхронный менеджер контекста. Внутри блока async with вы создаёте задачи через tg.create_task(...), а на выходе из блока группа дожидается их все. Ключевое отличие: если хотя бы одна задача падает с исключением (кроме CancelledError), группа отменяет все остальные и только потом выбрасывает ошибку наружу.

# Python 3.11+
import asyncio

async def fetch(name, delay, fail=False):
    try:
        await asyncio.sleep(delay)
    except asyncio.CancelledError:
        print(f"{name} отменён")
        raise
    if fail:
        raise ValueError(f"{name} упал")
    print(f"{name} завершился")
    return name

async def main_group():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fetch("a", 0.1, fail=True))
            tg.create_task(fetch("b", 0.5))
            tg.create_task(fetch("c", 0.5))
    except* ValueError as eg:
        # eg — ExceptionGroup, exceptions лежат в eg.exceptions
        for exc in eg.exceptions:
            print(f"Поймали: {exc}")

asyncio.run(main_group())

Теперь вывод будет другим: «a упал», затем «b отменён» и «c отменён», и только после этого сработает обработчик. Ни одна лишняя задача не доработала до конца. Это и есть структурная конкурентность: время жизни дочерних задач ограничено блоком, который их породил.

ExceptionGroup и синтаксис except*

Раз задач может упасть сразу несколько, TaskGroup собирает их исключения в ExceptionGroup — новый тип, тоже появившийся в 3.11. Ловят его через новый синтаксис except*, который разбирает группу по типам. Одна строка except* ValueError поймает все ValueError внутри группы, а не первое попавшееся.

Обычный except ValueError группу не поймает — это частая ошибка при переходе. Если вам нужно поймать любое исключение из группы целиком, пишите except* Exception либо ловите сам ExceptionGroup.

Что означает «структурная» в названии

Структурная конкурентность — принцип, при котором фоновая задача не может пережить область кода, в которой она создана.^* Открыли группу — все задачи внутри неё либо завершатся, либо будут отменены к моменту выхода из async with. Не бывает ситуации «функция вернула управление, а где-то в фоне ещё что-то крутится». Именно из-за этого исчезают предупреждения Task was destroyed but it is pending.

^* Термин из статьи Нейтана Смита о nurseries; в Python его популяризировала библиотека Trio, а стандартный asyncio перенял идею в 3.11.

Сравнение по существу

Обратите внимание на последнюю строку: gather удобно возвращает список значений в том же порядке, в каком переданы корутины. У TaskGroup такого возврата нет — вы сами держите ссылки на объекты Task и после блока читаете их .result().

Как забрать результаты из TaskGroup

# Python 3.11+
import asyncio

async def fetch(name, delay):
    await asyncio.sleep(delay)
    return f"результат {name}"

async def main():
    async with asyncio.TaskGroup() as tg:
        tasks = [
            tg.create_task(fetch(name, 0.1))
            for name in ("a", "b", "c")
        ]
    # сюда попадаем только после успешного завершения всех задач
    results = [t.result() for t in tasks]
    print(results)

asyncio.run(main())

Читать .result() нужно строго после выхода из блока — внутри задача может быть ещё не завершена. Если группа выбросила ExceptionGroup, до строки с results вы не дойдёте, что логично: результатов упавшего набора не бывает.

Когда что выбирать

• Берите TaskGroup, если задачи логически связаны и падение одной обесценивает остальные: параллельные запросы для одного ответа, набор записей в рамках одной операции, веерный опрос сервисов, где нужен полный набор.
• gather с return_exceptions=True уместен, когда задачи независимы и вам нужны результаты всех, включая упавшие: например, вы опрашиваете сто источников и готовы к тому, что часть недоступна, а частичный результат имеет ценность.
• Ни то ни другое, если задача должна жить дольше текущего блока — тогда это фоновый воркер, и им управляют явно, а не через группу, привязанную к области видимости.

Критерий короткий: нужна ли вам отмена соседей при первой ошибке. Нужна — TaskGroup. Не нужна и важен частичный результат — gather(return_exceptions=True).

Что ломается при переходе

• Старый except не ловит ошибку. После замены на TaskGroup ваш except ValueError перестанет срабатывать, потому что исключение теперь завёрнуто в ExceptionGroup. Меняйте на except*.
• CancelledError внутри задач. При отмене соседей каждая задача получит CancelledError. Если вы его перехватываете и не пробрасываете дальше (raise), задача притворится завершившейся и сломает механику группы. Ловите отмену только для очистки ресурсов и обязательно перевыбрасывайте.
• Ожидание порядка результатов. Код, который полагался на порядок списка от gather, нужно переписать на явное хранение ссылок на задачи.
• Версия рантайма. И TaskGroup, и ExceptionGroup, и синтаксис except* требуют Python 3.11. На 3.10 и раньше их нет; для более старых рантаймов существует внешняя библиотека с бэкпортом, но проверяйте её совместимость отдельно.

Что делать дальше

Пройдитесь по коду поиском asyncio.gather и разделите вызовы на две группы: где задачи связаны и где независимы. Первые переводите на TaskGroup, вторые оставляйте на gather(return_exceptions=True), но добавьте явную проверку результатов на экземпляры Exception, если её ещё нет.

После перехода включите в тестах сценарий, где одна из задач гарантированно падает, и убедитесь, что соседи получили CancelledError, а обработчик поймал ExceptionGroup. В проде смотрите на исчезновение предупреждений Task was destroyed but it is pending в логах — это самый прямой признак, что фоновые утечки закрылись. Точное поведение отмены сверяйте с разделом asyncio в документации вашей версии интерпретатора.