StackKit

Python asyncio in Practice: gather, Timeouts, Semaphores and TaskGroup

By · · 4 min read

asyncio lets one Python thread juggle many tasks that spend most of their time waiting, such as network requests, database calls or sleeping. While one task waits, the event loop runs another. It does not make CPU-heavy code faster; for that, use processes or a library that releases the GIL.

The examples below were run on Python 3.15. asyncio.timeout and asyncio.TaskGroup need Python 3.11 or newer.

Coroutines and the event loop

An async def function returns a coroutine. It does nothing until it is awaited or scheduled.

import asyncio

async def fetch(name, delay):
    await asyncio.sleep(delay)      # stands in for a network call
    return f"{name}:{delay}"

asyncio.run(fetch("a", 0.1))        # asyncio.run starts the event loop

Calling await fetch(...) three times in a row takes the sum of the delays. Running them concurrently takes about as long as the slowest.

Run tasks concurrently with gather

asyncio.gather runs awaitables together and returns their results in the order you passed them, not the order they finished.

async def main():
    results = await asyncio.gather(
        fetch("a", 0.3), fetch("b", 0.2), fetch("c", 0.1)
    )
    print(results)
# ['a:0.3', 'b:0.2', 'c:0.1']    total time: about 0.3 s, not 0.6 s

The three calls overlapped, so the total was close to the longest delay (0.3 s) instead of the 0.6 s sum.

Handling errors in gather

By default, if one awaitable raises, gather propagates the first exception, and the other tasks keep running in the background. To collect everything instead, pass return_exceptions=True:

async def bad():
    raise ValueError("bad")

res = await asyncio.gather(fetch("ok", 0.01), bad(), return_exceptions=True)
print(res)
# ['ok:0.01', ValueError('bad')]

Exceptions now appear as values in the list, so check each item with isinstance(item, Exception).

Timeouts with asyncio.timeout

Never wait indefinitely on I/O. Python 3.11 added a context manager:

try:
    async with asyncio.timeout(0.15):
        await fetch("slow", 1)
except TimeoutError:
    print("timed out")
# timed out

When the time limit passes, the awaiting task is cancelled and a TimeoutError is raised. For a single awaitable you can also write await asyncio.wait_for(coro, timeout=...). Always catch cancellation-related errors where you need to release resources, and re-raise asyncio.CancelledError rather than swallowing it.

Limit concurrency with a Semaphore

Starting 10,000 requests at once can overwhelm a server or exhaust file descriptors. A semaphore caps how many run at a time:

sem = asyncio.Semaphore(2)
active = peak = 0

async def job(i):
    global active, peak
    async with sem:
        active += 1
        peak = max(peak, active)
        await asyncio.sleep(0.05)
        active -= 1

await asyncio.gather(*(job(i) for i in range(6)))
print("peak concurrency:", peak)
# peak concurrency: 2

Six jobs were scheduled, but never more than two ran at once. For real HTTP work, set the limit to what the remote service tolerates.

Structured concurrency with TaskGroup

asyncio.TaskGroup (3.11+) is the modern way to run a set of tasks that belong together. Leaving the async with block waits for all of them, and if one fails, the others are cancelled.

async with asyncio.TaskGroup() as tg:
    t1 = tg.create_task(fetch("x", 0.01))
    t2 = tg.create_task(fetch("y", 0.02))

print(t1.result(), t2.result())
# x:0.01 y:0.02

Errors are collected into an ExceptionGroup, which you handle with except*. Compared with gather, a TaskGroup avoids "orphaned" tasks that keep running after a failure. Prefer it when you can require Python 3.11 or newer.

Mistakes to avoid

  • Blocking the loop. A call such as time.sleep(5) or a synchronous HTTP library stops every other task. Use await asyncio.sleep(), an async client, or await asyncio.to_thread(blocking_function).
  • Forgetting await. Calling a coroutine function without awaiting it creates a coroutine that never runs; Python warns "coroutine was never awaited".
  • Fire-and-forget tasks being garbage collected. If you use asyncio.create_task, keep a reference to the task (in a set or list), or it may be collected before it finishes.
  • Using asyncio for CPU work. Concurrency here means interleaving waiting, not parallel computation.
  • Calling asyncio.run inside a running loop, for example in a Jupyter notebook. There, use await directly.

When to use what

Need Tool
Run a known list of awaitables and collect results asyncio.gather
A group of tasks that should succeed or fail together asyncio.TaskGroup
Limit simultaneous operations asyncio.Semaphore
Bound the waiting time asyncio.timeout / wait_for
Run blocking code without freezing the loop asyncio.to_thread

Summary

gather overlaps waiting, timeout bounds it, a Semaphore limits it, and TaskGroup keeps related tasks together. Keep blocking calls out of the event loop, and you can handle thousands of I/O-bound operations from a single thread.

Keep reading