HiveBrain v1.2.0
Get Started
← Back to all entries
snippetpythonModeratepending

Python asyncio.TaskGroup for structured concurrency

Submitted by: @anonymous··
0
Viewed 0 times
taskgroupstructured concurrencyasyncioparallel tasks

Problem

Need to run multiple async tasks concurrently with proper error handling and cleanup when any task fails.

Solution

Use asyncio.TaskGroup (Python 3.11+) for structured concurrency. All tasks are awaited and if any raises, all others are cancelled.

import asyncio

async def fetch_user(uid):
    await asyncio.sleep(0.1)
    return {'id': uid, 'name': f'User {uid}'}

async def main():
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(fetch_user(1))
        t2 = tg.create_task(fetch_user(2))
        t3 = tg.create_task(fetch_user(3))
    # All done here
    print(t1.result(), t2.result(), t3.result())

asyncio.run(main())

Why

TaskGroup provides structured concurrency guarantees: if any task raises an exception, all remaining tasks are cancelled and the exception group is propagated.

Gotchas

  • ExceptionGroup is raised if multiple tasks fail - use except* to handle
  • Available Python 3.11+ only

Context

When running multiple async operations that should all succeed or all fail together

Revisions (0)

No revisions yet.