TaskGroup in Python

A nursery for tasks: it waits for all of them and cancels the rest on error.

import asyncio


async def fetch(name, delay):
    await asyncio.sleep(delay)
    return f"{name} done"


async def main():
    async with asyncio.TaskGroup() as group:
        first = group.create_task(fetch("basics", 0.02))
        second = group.create_task(fetch("traits", 0.01))
    print(first.result(), second.result())

    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(fetch("ok", 0.01))
            group.create_task(asyncio.sleep(0, result=None))
            raise RuntimeError("something broke")
    except* RuntimeError as errors:
        print("caught", [str(e) for e in errors.exceptions])


asyncio.run(main())

How it works

Every task started in the group is awaited at the exit.
One failure cancels the siblings and raises an ExceptionGroup.
It replaces the gather-plus-cleanup dance.

Keywords and builtins used here

as
async
await
def
except
fetch
for
main
print
raise
return
str
try
with

The run, in numbers

Lines: 24
Characters to type: 572
Tokens: 166
Three-star pace: 110 tpm

At the three-star pace of 110 tokens a minute, this run takes about 91 seconds.

Type this snippet

Step 1 of 4 in Async in depth, step 48 of 53 in Pythonic Python.

← Previous Next →