ПостStructured Concurrency в Python 3.11+ (TaskGroup vs Gather)

Structured Concurrency в Python 3.11+ (TaskGroup vs Gather) Если вы до сих пор используете asyncio.gather() для запуска конкурентных задач, вы, вероятно, теряете контроль над ошибками. Главная проблема gather: если одна таска падает, остальные продолжают работать (если не стоит return_exceptions=False, который убивает всё сразу, но "грязно"). Плюс, отлавливать несколько ошибок одновременно через try / except - это боль. В Python 3.11+ (и через trio / anyio раньше) стандартом стала Structured Concurrency через asyncio.TaskGroup. Как это выглядит: import asyncio async def task_fail(name, sec): await asyncio.sleep(sec) raise ValueError(f"Error in {name}") async def task_ok(name, sec): await asyncio.sleep(sec) print(f"Task {name} done") async def main(): try: async with asyncio.TaskGroup() as tg: tg.create_task(task_ok("A", 1)) tg.create_task(task_fail("B", 2)) # Упадет через 2 сек tg.create_task(task_fail("C", 1.5)) # Упадет через 1.5 сек except* ValueError as eg: # Обратите внимание на except* (ExceptionGroup) print(f"Caught errors: {eg.exceptions}") # Запуск # asyncio.run(main()) Почему TaskGroup это выбор Middle+: 1. Атомарность скоупа: Если одна задача внутри контекстного менеджера падает, TaskGroup автоматически отменяет (cancel) все остальные запущенные задачи в группе. Вы не оставляете "зомби-процессы" в фоне. 2. Exception Groups: Если упало несколько задач одновременно (или во время отмены), Python соберет их в ExceptionGroup. Конструкция except* позволяет элегантно обрабатывать дерево исключений. Вывод: Для скриптов gather сойдет. Для надежных сервисов, где важна консистентность состояния и корректный шатдаун корутин - переезжайте на TaskGroup. #asyncio #python311 #bestpractices #concurrency 📲 Мы в MAX 👉@BookPython