asyncio.TaskGroup: что происходит с соседями при ошибке

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

Связать задачи одним владельцем

TaskGroup создаёт группу дочерних задач внутри async with. Выход из блока дожидается их завершения. Когда одна задача завершается исключением, отличным от CancelledError, группа отменяет оставшиеся задачи и ждёт их остановки. После этого ошибки выдаются в ExceptionGroup. Этот механизм удобен для частей одной операции, которые не должны жить отдельно от неё.

Отмена доставляется в точке приостановки, поэтому задача обязана освобождать ресурсы через finally и не проглатывать CancelledError. После очистки отмена должна продолжить распространение. Синхронный бесконечный цикл без await не получает управление от цикла событий и не становится управляемым только от помещения в TaskGroup.

Проверить порядок через события

В примере медленная задача сообщает, что начала работу, и ждёт события, которое никто не установит. Ошибающаяся задача ждёт её старта и затем поднимает ValueError. Поэтому тест не опирается на случайную длительность sleep. Группа должна отменить ожидание, выполнить finally и только затем вернуть управление обработчику ошибки.

Для запуска нужен Python 3.11 или новее. except* обрабатывает исключения внутри группы. Проверяем наличие ValueError и факт завершения очистки. Это точнее, чем наблюдать, что программа просто перестала печатать сообщения.

import asyncio

async def main():
    started = asyncio.Event()
    cleaned = []
    async def waiting_part():
        try:
            started.set()
            await asyncio.Event().wait()
        finally:
            cleaned.append("released")
    async def failing_part():
        await started.wait()
        raise ValueError("invalid report input")
    errors = []
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(waiting_part())
            group.create_task(failing_part())
    except* ValueError as error_group:
        errors.extend(error_group.exceptions)
    assert len(errors) == 1
    assert str(errors[0]) == "invalid report input"
    assert cleaned == ["released"]
    print("failure collected after cleanup")

asyncio.run(main())

Где провести границу группы

Объединяйте задачи по смыслу результата. Если сбой подготовки одной части делает весь отчёт непригодным, общая группа подходит. Если это независимые задания, которые должны закончиться каждый со своим статусом, перехватите ожидаемые ошибки внутри задания и верните явный результат. Тогда ошибка одного элемента не будет автоматически отменять остальные.

Проверьте также вызывающий код: он должен получать готовые значения после выхода из группы, когда дочерние задачи уже завершены. Сохранённая где-то отдельно ссылка на задачу не делает её владельцем. Ясная граница помогает не оставлять фоновую работу после завершения основной операции.

Опора на базовую проверку поведения

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