Introduction

Python's asyncio module provides three primary tools for running multiple coroutines concurrently: asyncio.gather, asyncio.wait, and asyncio.TaskGroup. Each serves a different use case. gather is the simplest and returns ordered results. wait gives you fine-grained control over completion conditions. TaskGroup (Python 3.11+) enforces structured concurrency where all tasks are guaranteed to finish before the context exits.

Choosing the wrong one leads to either lost exceptions, orphaned tasks, or unnecessarily complex code. This article breaks down exactly when to use each.

asyncio.gather: Simple Aggregation

asyncio.gather takes multiple coroutines or futures, runs them concurrently, and returns a list of results in the same order they were passed in.

1import asyncio
2
3async def fetch_user(user_id):
4    await asyncio.sleep(0.5)  # simulate network call
5    return {"id": user_id, "name": f"User {user_id}"}
6
7async def main():
8    results = await asyncio.gather(
9        fetch_user(1),
10        fetch_user(2),
11        fetch_user(3),
12    )
13    for user in results:
14        print(user)
15
16asyncio.run(main())

The results list always matches the input order, regardless of which coroutine finished first. This makes gather ideal for "fan-out, collect all" patterns like fetching multiple API endpoints in parallel.

Exception Behavior

By default, if any coroutine raises an exception, gather propagates it immediately. The other coroutines continue running in the background, but their results are lost:

1async def failing_task():
2    raise ValueError("something broke")
3
4async def main():
5    try:
6        results = await asyncio.gather(
7            fetch_user(1),
8            failing_task(),
9            fetch_user(3),
10        )
11    except ValueError as e:
12        print(f"Caught: {e}")
13        # results from fetch_user(1) and fetch_user(3) are NOT available
14
15asyncio.run(main())

Setting return_exceptions=True changes the behavior: exceptions are returned as values in the results list instead of being raised:

1results = await asyncio.gather(
2    fetch_user(1),
3    failing_task(),
4    fetch_user(3),
5    return_exceptions=True,
6)
7# results[0] = {"id": 1, "name": "User 1"}
8# results[1] = ValueError("something broke")
9# results[2] = {"id": 3, "name": "User 3"}

asyncio.wait: Fine-Grained Control

asyncio.wait returns two sets of tasks (done and pending) based on a completion condition. It accepts asyncio.Task objects or futures, not bare coroutines (in Python 3.11+ bare coroutines raise a deprecation warning).

1import asyncio
2
3async def fetch_price(exchange, symbol):
4    delay = {"binance": 0.3, "coinbase": 0.8, "kraken": 1.2}
5    await asyncio.sleep(delay.get(exchange, 1.0))
6    return {"exchange": exchange, "symbol": symbol, "price": 42000.0}
7
8async def main():
9    tasks = [
10        asyncio.create_task(fetch_price("binance", "BTC")),
11        asyncio.create_task(fetch_price("coinbase", "BTC")),
12        asyncio.create_task(fetch_price("kraken", "BTC")),
13    ]
14
15    # Return as soon as the first exchange responds
16    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
17
18    for task in done:
19        print("First result:", task.result())
20
21    # Cancel the rest since we only needed one
22    for task in pending:
23        task.cancel()
24
25asyncio.run(main())

Completion Conditions

asyncio.wait also accepts a timeout parameter. If the timeout expires, whatever has completed goes into done and everything else goes into pending:

done, pending = await asyncio.wait(tasks, timeout=2.0)
print(f"{len(done)} completed, {len(pending)} still running")

Key Difference from gather

asyncio.wait does not raise exceptions automatically. You must call task.result() on each completed task to retrieve the result or re-raise the exception. This gives you full control but requires more careful handling.

asyncio.TaskGroup: Structured Concurrency

Introduced in Python 3.11, TaskGroup enforces a critical guarantee: all tasks created within the group are awaited before the async with block exits. If any task raises an exception, all remaining tasks are cancelled and the exception is re-raised as an ExceptionGroup.

1import asyncio
2
3async def process_order(order_id):
4    await asyncio.sleep(0.5)
5    print(f"Order {order_id} processed")
6    return order_id
7
8async def main():
9    async with asyncio.TaskGroup() as tg:
10        task1 = tg.create_task(process_order(101))
11        task2 = tg.create_task(process_order(102))
12        task3 = tg.create_task(process_order(103))
13
14    # All tasks are guaranteed complete here
15    print(task1.result(), task2.result(), task3.result())
16
17asyncio.run(main())

Exception Handling with ExceptionGroup

When a task inside a TaskGroup fails, the group cancels all other tasks, waits for cancellation to finish, then raises an ExceptionGroup:

1async def risky_task(task_id):
2    if task_id == 2:
3        raise RuntimeError(f"Task {task_id} failed")
4    await asyncio.sleep(0.5)
5    return task_id
6
7async def main():
8    try:
9        async with asyncio.TaskGroup() as tg:
10            tg.create_task(risky_task(1))
11            tg.create_task(risky_task(2))
12            tg.create_task(risky_task(3))
13    except* RuntimeError as eg:
14        for exc in eg.exceptions:
15            print(f"Caught: {exc}")
16
17asyncio.run(main())

The except* syntax (Python 3.11+) matches specific exception types within the group. This prevents the common problem with gather where one failure silently orphans other tasks.

Comparison Table

Decision Guide

Use asyncio.gather when you need all results in order and failures are either unlikely or acceptable to handle with return_exceptions=True. This covers the majority of "call N endpoints in parallel" scenarios.

Use asyncio.wait when you need partial results, race semantics (first to respond wins), or timeout-based completion. It is the right tool for health checks, competitive fetches, or progress monitoring.

Use asyncio.TaskGroup when task cleanup matters. Any code that creates background tasks inside a request handler, a transaction, or a resource scope should use TaskGroup to prevent orphaned coroutines that leak memory or database connections.

Common Pitfalls

• Passing bare coroutines to asyncio.wait instead of wrapping them with asyncio.create_task(). In Python 3.11+ this produces a deprecation warning, and in future versions it will be an error.
• Using asyncio.gather without return_exceptions=True and losing results from tasks that completed successfully when one task fails.
• Forgetting to cancel pending tasks after asyncio.wait with FIRST_COMPLETED. The pending tasks keep running in the background, consuming resources.
• Catching Exception instead of using except* with TaskGroup. An ExceptionGroup is not a single exception, and a bare except Exception will catch it but lose the structured information.
• Using TaskGroup on Python versions below 3.11. The exceptiongroup backport package exists, but TaskGroup itself requires 3.11+.

Summary

• asyncio.gather is the default choice for running coroutines in parallel and collecting ordered results. Use return_exceptions=True when partial success is acceptable.
• asyncio.wait provides control over completion conditions (first done, first exception, timeout). Always cancel pending tasks when you no longer need them.
• asyncio.TaskGroup is the safest option for structured concurrency. It guarantees no orphaned tasks and uses ExceptionGroup for clean error reporting.
• For new code targeting Python 3.11+, prefer TaskGroup over gather whenever task lifecycle management matters.