Асинхронное программирование в Python: от простого к сложному. Часть 4 Сегодня поговорим подробнее о Task. Разберем TaskGroup (новинка Python 3.11), групповую обработку ошибок и полезные приемы при работе с задачами TaskGroup - современный способ управления группой связанных задач. Главная особенность - встроенная обработка ошибок через ExceptionGroup: async def process_user(user_id: int): # Имитация запросов к разным сервисам if user_id % 3 == 0: raise ValueError(f"Ошибка получения данных для пользователя {user_id}") if user_id % 5 == 0: raise ConnectionError(f"Ошибка соединения для пользователя {user_id}") await asyncio.sleep(1) return f"Данные пользователя {user_id}" async def main(): try: async with asyncio.TaskGroup() as tg: # Создаем группу задач tasks = [ tg.create_task(process_user(i), name=f"user_{i}") for i in range(10) ] except* ValueError as eg: # Ловим все ошибки типа ValueError print(f"Ошибки получения данных: {eg.exceptions}") except* ConnectionError as eg: # Отдельно обрабатываем ошибки соединения print(f"Ошибки соединения: {eg.exceptions}") except* Exception as eg: # Все остальные ошибки print(f"Непредвиденные ошибки: {eg.exceptions}") Что такое except* и как это работает: - Обычный except ловит только одно исключение - except* может поймать несколько исключений одного типа - Исключения группируются в ExceptionGroup - При выходе из контекста TaskGroup все задачи автоматически отменяются Преимущества TaskGroup: - Автоматическая отмена всех задач при ошибке - Группировка исключений Полезные приемы при работе с задачами: 1. Добавление callback для реакции на завершение: def log_result(task): try: result = task.result() print(f"✅ Задача {task.get_name()} вернула: {result}") except Exception as e: print(f"❌ Задача {task.get_name()} упала с ошибкой: {e}") async def main(): async with asyncio.TaskGroup() as tg: task = tg.create_task(process_user(1), name="user_processor") task.add_done_callback(log_result) 2. Получение информации о задаче: # Создаем задачу task = asyncio.create_task(some_coro(), name="important_task") # Доступная информация print(task.get_name()) # Имя задачи print(task.done()) # Завершена ли print(task.cancelled()) # Отменена ли print(task.exception()) # Исключение (если есть) 3. Таймаут для группы задач: async def main(): try: async with asyncio.timeout(5): async with asyncio.TaskGroup() as tg: tasks = [ tg.create_task(process_user(i)) for i in range(10) ] except asyncio.TimeoutError: print("Превышено время выполнения группы задач") 4. Ожидание первого успешного результата: async def fetch_with_fallback(urls: list[str]): async def fetch_url(url: str): async with aiohttp.ClientSession() as session: async with session.get(url) as response: if response.status == 200: return await response.json() raise ValueError(f"Ошибка получения данных: {response.status}") try: async with asyncio.TaskGroup() as tg: tasks = [ tg.create_task(fetch_url(url), name=f"fetch_{i}") for i, url in enumerate(urls) ] # Ждем первый успешный результат for task in tasks: try: return await task except ValueError: continue except* Exception as eg: raise ValueError("Все URL недоступны") Практические кейсы использования TaskGroup: - Параллельные запросы к нескольким API - Выполнение связанных операций с разными сервисами В следующей части поговорим про очереди и потоки данных Поддержать и посмотреть уникальные видео на Boosty Посмотреть на Youtube