Python asyncio: Do Many Waits at Once

Most programs spend most of their time waiting: for a web server to answer, a database to return rows, or a file to arrive. While one wait is in progress, the CPU sits idle. Python’s asyncio lets a single program start many of those waits and handle each answer as it arrives, so ten network calls take about as long as the slowest one instead of all of them added up.

In the Python concurrency overview, we looked at threads and processes. This article covers the third option: asyncio, which has been part of Python since version 3.4. The examples here use features from Python 3.11 and newer.

When to use asyncio

Asyncio shines when your program is I/O bound: it waits on networks, disks or other programs. Think web scrapers, API clients, chat servers and anything that talks to many services at once. It does not make CPU-heavy work faster; for number crunching, use multiprocessing instead. If you’re unsure about the difference, read about I/O bound vs CPU bound first.

Your first coroutine

A function defined with async def is a coroutine function. Inside it, await marks a point where the function may pause while it waits. asyncio.run() starts the event loop and runs a coroutine to completion:

import asyncio


async def greet(name):
    await asyncio.sleep(1)
    return f"Hello, {name}!"


print(asyncio.run(greet("Python Land")))Code language: Python (python)

Hello, Python Land!Code language: plaintext (plaintext)

asyncio.sleep() is the async version of time.sleep(). It stands in for a real wait, like a network call, throughout this article.

Calling is not running

Here’s the first thing that surprises everyone. Calling a coroutine function does not run it. It creates a coroutine object, and nothing happens until something awaits it:

import asyncio


async def greet(name):
    await asyncio.sleep(1)
    return f"Hello, {name}!"


coroutine = greet("world")
print(coroutine)
print(asyncio.run(coroutine))Code language: Python (python)

<coroutine object greet at 0x104f1e5c0>
Hello, world!Code language: plaintext (plaintext)

If you ever see a warning like RuntimeWarning: coroutine 'greet' was never awaited, you called a coroutine function and forgot the await.

Want to try this with instant feedback? Run your first coroutine in our free interactive lesson, no setup needed.

Running waits side by side with gather

Awaiting coroutines one after another still runs them one at a time. To let them overlap, hand them to asyncio.gather():

import asyncio
import time


async def check(name, seconds):
    await asyncio.sleep(seconds)  # stands in for a network call
    return f"{name} answered after {seconds}s"


async def main():
    start = time.perf_counter()
    results = await asyncio.gather(
        check("database", 1),
        check("web server", 2),
        check("cache", 1.5),
    )
    for line in results:
        print(line)
    print(f"Total: {time.perf_counter() - start:.1f}s")


asyncio.run(main())Code language: Python (python)

database answered after 1s
web server answered after 2s
cache answered after 1.5s
Total: 2.0sCode language: plaintext (plaintext)

The three waits add up to 4.5 seconds, but the program finishes in 2: the time of the slowest one. gather() returns the results in the order you passed the coroutines, not the order in which they finished.

When one of them fails

By default, if one coroutine raises an exception, gather() raises it too, and you lose the other results. Pass return_exceptions=True to get exceptions back as values instead:

import asyncio


async def check(name, seconds):
    await asyncio.sleep(seconds)
    if name == "cache":
        raise ConnectionError("cache is down")
    return name


async def main():
    results = await asyncio.gather(
        check("database", 1),
        check("cache", 0.5),
        return_exceptions=True,
    )
    print(results)


asyncio.run(main())Code language: Python (python)

['database', ConnectionError('cache is down')]Code language: plaintext (plaintext)

TaskGroup: the modern way

Since Python 3.11, asyncio.TaskGroup is the recommended way to run tasks together. Every task you create inside the async with block runs concurrently, and the block only ends when all of them are done:

import asyncio


async def check(name, seconds):
    await asyncio.sleep(seconds)
    return f"{name} answered after {seconds}s"


async def main():
    async with asyncio.TaskGroup() as group:
        db = group.create_task(check("database", 1))
        web = group.create_task(check("web server", 2))
    print(db.result())
    print(web.result())


asyncio.run(main())Code language: Python (python)

The big advantage over gather() is what happens on failure: if one task raises, the TaskGroup cancels the others and raises all errors together as an ExceptionGroup. No task is left running in the background by accident.

To catch errors from an ExceptionGroup, use except* (also new in Python 3.11). It matches the errors of a given type inside the group:

import asyncio


async def check(name, seconds):
    await asyncio.sleep(seconds)
    if name == "web server":
        raise ConnectionError(f"{name} is down")
    return f"{name} answered after {seconds}s"


async def main():
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(check("database", 1))
            group.create_task(check("web server", 2))
    except* ConnectionError as group_error:
        for error in group_error.exceptions:
            print("Failed:", error)


asyncio.run(main())Code language: Python (python)

Failed: web server is downCode language: plaintext (plaintext)

Even when only one task fails, you get a group, so loop over group_error.exceptions to see each error.

Put a deadline on every wait

A network call that never answers will make your coroutine wait forever. asyncio.timeout() puts a limit on everything inside its block:

import asyncio


async def slow_service():
    await asyncio.sleep(10)
    return "finally!"


async def main():
    try:
        async with asyncio.timeout(2):
            print(await slow_service())
    except TimeoutError:
        print("Gave up after 2 seconds")


asyncio.run(main())Code language: Python (python)

Gave up after 2 secondsCode language: plaintext (plaintext)

When the time is up, asyncio cancels the waiting coroutine and raises TimeoutError. On Python versions before 3.11, use asyncio.wait_for(slow_service(), timeout=2) instead.

Real HTTP requests with httpx

asyncio.sleep() is fine for learning, but real programs await real I/O. The popular requests library is synchronous, so it can’t be awaited. Use httpx instead, which has an async client (python -m pip install httpx):

import asyncio
import httpx


async def fetch_status(client, url):
    response = await client.get(url)
    return url, response.status_code


async def main():
    urls = [
        "https://httpbin.org/status/200",
        "https://httpbin.org/status/404",
        "https://httpbin.org/delay/1",
    ]
    async with httpx.AsyncClient(timeout=10) as client:
        results = await asyncio.gather(*(fetch_status(client, url) for url in urls))
    for url, status in results:
        print(status, url)


asyncio.run(main())Code language: Python (python)

200 https://httpbin.org/status/200
404 https://httpbin.org/status/404
200 https://httpbin.org/delay/1Code language: plaintext (plaintext)

Create one AsyncClient and share it between requests: it reuses connections, and its timeout applies to every request. If you know requests, the API will feel familiar.

Never block the event loop

Asyncio runs all your coroutines in a single thread, and it can only switch between them at an await. A normal blocking call, like time.sleep(), requests.get() or a slow computation, freezes everything until it returns. That’s the most common asyncio bug.

When you must call blocking code, hand it to a thread with asyncio.to_thread(), and the event loop keeps running:

import asyncio
import time


def blocking_work():
    time.sleep(2)  # a library call that does not know about asyncio
    return "blocking work done"


async def heartbeat():
    for _ in range(4):
        print("tick")
        await asyncio.sleep(0.5)


async def main():
    result, _ = await asyncio.gather(
        asyncio.to_thread(blocking_work),
        heartbeat(),
    )
    print(result)


asyncio.run(main())Code language: Python (python)

tick
tick
tick
tick
blocking work doneCode language: plaintext (plaintext)

The heartbeat keeps ticking while the blocking work runs. Call blocking_work() directly inside a coroutine instead, and the ticks stop for two seconds.

asyncio vs threads vs processes

• asyncio: thousands of concurrent waits in one thread, with explicit switch points at every await. Best for network-heavy code, but you need async libraries.
• Threads: work with any blocking library, but are heavier, and shared data needs locks.
• Processes: the classic way around the GIL for CPU-heavy work. With the free-threaded build of Python (officially supported since 3.14, but still optional), threads can run CPU-heavy work in parallel too.

Keep learning

You can now write coroutines, run waits side by side with gather() and TaskGroup, put a deadline on every wait, make async HTTP requests, and keep blocking code out of the event loop. Our free Async Python course builds on exactly this: you build Pulse, a tested command-line health checker that checks every target at once, then learn how the event loop works and add tasks, cancellation, queues and async testing. The asyncio documentation is the full reference.