Running Many at Once

asyncio.gather returns results in argument order; TaskGroup cancels the siblings when one task fails.

Syntaxresults = await asyncio.gather(*aws, return_exceptions=False)

await on a coroutine runs it to completion before the next line. To get two things happening at once, both have to be handed to the loop before you wait on either. That is what asyncio.gather and asyncio.TaskGroup do.

gather

asyncio.gather(*aws, return_exceptions=False) schedules everything, waits for all of it, and returns a list of results in argument order — never completion order. Bare coroutines are wrapped in tasks for you, which is why the explicit version below behaves identically:

tasks = [asyncio.create_task(load(k)) for k in keys]
results = [await t for t in tasks]      # all started before the first await

results = await asyncio.gather(*(load(k) for k in keys))   # the same thing

What happens when one fails

By default the first exception propagates out of gather immediately, and the other awaitables keep running, unsupervised, their results discarded. With return_exceptions=True nothing is raised: the exception object sits in the results list in that awaitable's position, and it is your job to check for it.

TaskGroup (3.11+)

async with asyncio.TaskGroup() as tg: plus tg.create_task(...) waits for every task at the end of the block. If one raises, the siblings are cancelled, and the block raises an ExceptionGroup that you unpack with except*. Nothing escapes the block still running, which is what makes it the better default; reach for gather when you want a plain ordered list of results from a fixed set.

Example

Try it yourself
Loading editor…
Press Run to execute the code.

When to use it

  • A dashboard endpoint that needs three internal services before it can render.
  • Refreshing 500 tenant caches overnight where a few tenants are expected to fail.
  • Racing two read replicas and taking whichever answers first.
  • Replacing a for loop of awaits that is slow only because it is serial.

More examples

TaskGroup with named results

Three independent calls, one deadline-free block, named results instead of positional ones. Because .result() is only read after the block exits, it can never raise InvalidStateError. Use this shape in a request handler: if billing is down, the other two calls are cancelled rather than left running after the handler has already failed.

Example · python
import asyncio

import httpx

async def build_dashboard(client, uid):
    async with asyncio.TaskGroup() as tg:
        # create_task returns the Task, so you can name each result
        user = tg.create_task(client.get(f"/users/{uid}"))
        orders = tg.create_task(client.get(f"/users/{uid}/orders"))
        billing = tg.create_task(client.get(f"/users/{uid}/billing"))
    # past the end of the block every task is done, or we never got here
    return {
        "user": user.result().json(),
        "orders": orders.result().json(),
        "billing": billing.result().json(),
    }

# the caller unpacks an ExceptionGroup, not a bare exception
async def main():
    async with httpx.AsyncClient(base_url="https://api.example.com") as client:
        try:
            print(await build_dashboard(client, 42))
        except* httpx.HTTPError as group:
            print("upstream failed:", [str(e) for e in group.exceptions])

Partial failure you are allowed to tolerate

The one case where return_exceptions=True is right: a batch job where each item stands alone and the report matters more than failing fast. Note the isinstance check against BaseException rather than Exception, so a CancelledError in the list is not mistaken for a successful result.

Example · python
import asyncio
import logging

async def refresh_all(client, shop_ids):
    results = await asyncio.gather(
        *(refresh(client, sid) for sid in shop_ids),
        return_exceptions=True,       # one bad shop must not kill the batch
    )
    ok, failed = {}, {}
    # zip keeps each result lined up with the id that produced it
    for sid, result in zip(shop_ids, results):
        if isinstance(result, BaseException):
            failed[sid] = result      # BaseException, so cancellations count too
            logging.warning("shop %s failed: %r", sid, result)
        else:
            ok[sid] = result
    return ok, failed

First answer wins, with as_completed

as_completed(fs, *, timeout=None) yields awaitables in completion order, which is what you want for a race rather than a batch. The finally block is not optional: neither as_completed nor an early return cancels the losers, so without it the slow mirrors keep downloading after you have answered.

Example · python
import asyncio

async def resolve(mirrors, path):
    tasks = [asyncio.create_task(fetch(m + path)) for m in mirrors]
    try:
        # 3.13+: as_completed is an async iterator; plain for still works too
        async for finished in asyncio.as_completed(tasks):
            try:
                return await finished      # the first mirror that answers
            except OSError:
                continue                   # try the next one to finish
        raise RuntimeError("every mirror failed")
    finally:
        for t in tasks:
            t.cancel()                     # as_completed cancels nothing itself

Related

py async basicspy async httppy async timeoutspy async pitfallspy async queue

Discussion

  • Be the first to comment on this lesson.