TaskGroup manages a set of tasks with automatic cleanup.
1import asyncio23async def fetch(url):4 await asyncio.sleep(1)5 return f"Data from {url}"67async def main():8 # Python 3.11+9 async with asyncio.TaskGroup() as tg:10 task1 = tg.create_task(fetch("url1"))11 task2 = tg.create_task(fetch("url2"))12 task3 = tg.create_task(fetch("url3"))1314 # All tasks completed15 print(task1.result())16 print(task2.result())17 print(task3.result())1819# With error handling20async def unreliable_fetch(url):21 await asyncio.sleep(1)22 if "error" in url:23 raise ValueError(f"Failed: {url}")24 return f"Data from {url}"2526async def main_with_errors():27 try:28 async with asyncio.TaskGroup() as tg:29 task1 = tg.create_task(unreliable_fetch("url1"))30 task2 = tg.create_task(unreliable_fetch("error"))31 except* ValueError as eg:32 for exc in eg.exceptions:33 print(f"Error: {exc}")3435asyncio.run(main())
Benefits: