Ghostly Python
📱 Фишка: asyncio.TaskGroup — надёжное управление группами задач
TaskGroup автоматически следит за всеми запущенными в нём задачами. Если любая задача падает с ошибкой — отменяются все остальные, и вы точно ничего не пропустите. Удобнее и безопаснее, чем ручной gather/wait.
import asyncio
async def worker(name, delay):
print(f"Задача {name} стартует")
await asyncio.sleep(delay)
if name == "fail":
raise ValueError("Ошибка в задаче!")
print(f"Задача {name} завершена")
return f"Результат {name}"
async def main():
try:
async with asyncio.TaskGroup() as tg:
task1 = tg.create_task(worker("A", 1))
task2 = tg.create_task(worker("B", 2))
task3 = tg.create_task(worker("fail", 0.5))
task4 = tg.create_task(worker("C", 1.5))
# сюда мы попадём, только если все задачи успешны
print("Все задачи завершены без ошибок")
except Exception as e:
print(f"Перехвачено исключение: {e}")
# все незавершённые задачи уже отменены автоматически
asyncio.run(main())
📌 Как это работает?
— При входе в async with создаётся группа задач.
— Как только одна задача вызывает исключение — группа немедленно отменяет все остальные.
— После выхода из блока можно быть уверенным: либо все задачи выполнились успешно, либо ни одна «незакрытая» ошибка не осталась незамеченной.
🐍 Ghostly Python | #фишки
🔥1
22 · 3K ·