asyncio and Asynchronous HTTP

You convert a slow script to async def, add await in front of the calls, and run it. It is exactly as slow as before, or slower. Nothing in the syntax told you why, because the reason is not in the syntax: a blocking call inside a coroutine stops every other task in the program, and the interpreter will not warn you about it.

asyncio does not give Python a second thread. It gives you a way to write down the places where your code is going to wait, so the interpreter can run something else during those moments. When you lie about where those places are, the whole model collapses and you are left with an event loop that runs your blocking function one call at a time.

The event loop is a scheduler with one thread

An event loop keeps a set of runnable callbacks and a set of things it is waiting for, which on Unix are file descriptors registered with select, poll or epoll. It runs whatever is runnable until that work voluntarily gives up control, then asks the operating system which file descriptors are ready, then runs the callbacks for those.

One thread runs all of it. That is the whole design, and it has a direct consequence: concurrency without parallelism. Two coroutines can both be in flight at the same moment, but only one instruction of Python is executing at any instant. If the workload is arithmetic, asyncio gives you nothing that threads did not. If the workload is waiting on a socket, asyncio gives you thousands of concurrent waits from one thread, because waiting costs a file descriptor rather than a thread stack.

The unit of scheduling is the task, created with asyncio.create_task or by the helpers in the next sections. Tasks start running when the loop next gets control, not when you create them. A coroutine object you build and never await is not scheduled and does nothing, and Python prints a warning about it at interpreter exit rather than silently ignoring it.

await marks the places where your function can be suspended

An async def function returns a coroutine object when you call it. It does no work until something awaits it or wraps it in a task. Inside, await means "this may take a while, so the loop is free to run other tasks until it finishes".

import asyncio


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


async def main():
    a = asyncio.create_task(fetch("a", 1))
    b = asyncio.create_task(fetch("b", 0.5))
    print(await a)
    print(await b)


asyncio.run(main())

start a
start b
done  b
done  a
a:1
b:0.5

Both tasks started before either finished, and b finished first because it slept less. The prints appear in completion order, but the values come back in the order you awaited them, which is the distinction that matters: scheduling order and result order are separate things.

If you had written await fetch("a", 1) followed by await fetch("b", 0.5), you would get the same output minus the interleaving, and 1.5 seconds of wall clock instead of 1.0. Sequential awaits are the single most common reason a converted script is not faster.

A blocking call inside a coroutine stalls everything

await only yields where the coroutine actually reaches an await that suspends. A call to time.sleep, a synchronous HTTP client, or a heavy database driver holds the thread, and the loop cannot run anything else until it returns. Nothing is scheduled during that time, including timeouts, heartbeats and other requests.

import asyncio
import time

import httpx

BASE = "http://127.0.0.1:8801"
NAMES = [f"item-{n}" for n in range(1, 9)]


async def blocking_fetch(client, name):
    # time.sleep blocks the thread, and the event loop is that thread
    time.sleep(0.3)
    response = await client.get("/item", params={"name": name})
    return response.json()["name"]


async def main():
    async with httpx.AsyncClient(base_url=BASE, timeout=5.0) as client:
        started = time.perf_counter()
        names = await asyncio.gather(*(blocking_fetch(client, name) for name in NAMES))
        print(f"gather over blocking_fetch  {time.perf_counter() - started:5.2f}s")
    print(names[0], names[-1])


asyncio.run(main())

gather over blocking_fetch   2.86s  item-1 ... item-8

Eight requests gathered concurrently, and it took longer than running them one at a time would have. gather did its job: it put eight tasks on the loop. Each one then blocked the loop for 0.3 seconds, so the requests could not overlap, and the whole thing cost 2.4 seconds of sleeping plus the request time on top.

When you cannot replace the blocking call, move it to a thread with asyncio.to_thread, which is a thin wrapper over loop.run_in_executor that uses the default executor:

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

import httpx

BASE = "http://127.0.0.1:8801"
NAMES = [f"item-{n}" for n in range(1, 9)]


def legacy_fetch(name):
    """A blocking call you do not control: an old client, a driver, a CLI."""
    with httpx.Client(base_url=BASE, timeout=5.0) as client:
        time.sleep(0.3)
        return client.get("/item", params={"name": name}).json()["name"]


async def main():
    started = time.perf_counter()
    names = await asyncio.gather(*(asyncio.to_thread(legacy_fetch, name) for name in NAMES))
    print(f"asyncio.to_thread   {time.perf_counter() - started:5.2f}s  {names[0]} ... {names[-1]}")

    loop = asyncio.get_running_loop()
    started = time.perf_counter()
    with ThreadPoolExecutor(max_workers=8) as pool:
        names = await asyncio.gather(
            *(loop.run_in_executor(pool, legacy_fetch, name) for name in NAMES)
        )
    print(f"run_in_executor     {time.perf_counter() - started:5.2f}s  {names[0]} ... {names[-1]}")


asyncio.run(main())

asyncio.to_thread    0.64s  item-1 ... item-8
run_in_executor      0.62s  item-1 ... item-8

Both recover the concurrency. to_thread uses a default executor whose size you do not control, and run_in_executor with an executor you create lets you cap it, which matters when each blocked thread holds a connection. The ceiling is real: eight threads are eight threads, and this is how you end up with an asyncio program that is secretly a thread pool.

gather and TaskGroup collect results with different failure behaviour

asyncio.gather takes awaitables and returns a list of results in the order you passed them. By default the first exception propagates and the other tasks keep running in the background, which is rarely what you want. return_exceptions=True changes that to a list where each entry is either a result or an exception object, and nothing is raised.

import asyncio


async def work(name, fail=False):
    await asyncio.sleep(0.15)
    if fail:
        raise ValueError(f"{name} failed")
    return name


async def with_gather():
    results = await asyncio.gather(
        work("a"), work("b", fail=True), work("c"), return_exceptions=True
    )
    for result in results:
        print(f"  gather result: {result!r}")


async def with_task_group():
    try:
        async with asyncio.TaskGroup() as group:
            for name, fail in (("a", False), ("b", True), ("c", False)):
                group.create_task(work(name, fail), name=name)
    except* ValueError as group:
        for error in group.exceptions:
            print(f"  task group error: {type(error).__name__}: {error}")
    else:
        print("  task group finished without error")


async def main():
    print("gather with return_exceptions=True")
    await with_gather()
    print("TaskGroup with a failing task")
    await with_task_group()


asyncio.run(main())

gather with return_exceptions=True
  gather result: 'a'
  gather result: ValueError('b failed')
  gather result: 'c'
TaskGroup with a failing task
  task group error: ValueError: b failed

asyncio.TaskGroup, added in 3.11, behaves differently in the way that matters for a service: when one task fails, the group cancels the others and waits for them to finish unwinding before it raises.

import asyncio

finished = []


async def work(name, seconds, fail=False):
    try:
        await asyncio.sleep(seconds)
    except asyncio.CancelledError:
        finished.append(f"{name}:cancelled")
        raise
    finished.append(f"{name}:done")
    if fail:
        raise ValueError(f"{name} failed")


async def main():
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(work("fast-fail", 0.05, fail=True))
            group.create_task(work("slow-a", 1.0))
            group.create_task(work("slow-b", 1.0))
    except* ValueError as group:
        print("errors:", [str(e) for e in group.exceptions])
    print("tasks:", finished)


asyncio.run(main())

errors: ['fast-fail failed']
tasks: ['fast-fail:done', 'slow-a:cancelled', 'slow-b:cancelled']

The two slow tasks never completed, and their CancelledError handlers ran, which is your chance to close a connection or write a partial result. This is the behaviour you want in a service, because a failed batch should not leave work running after the caller has been told about the failure. Use gather when you genuinely want every result regardless of individual failures, and TaskGroup when a failure should stop the batch.

Timeouts and cancellation are the same mechanism

asyncio.timeout sets a deadline on a block. When the deadline passes, the loop cancels the tasks inside it and the async with raises TimeoutError, which is the builtin one, not asyncio.TimeoutError.

import asyncio

import httpx

BASE = "http://127.0.0.1:8801"


async def fetch(client, name):
    response = await client.get("/item", params={"name": name})
    response.raise_for_status()
    return response.json()["name"]


async def main():
    async with httpx.AsyncClient(base_url=BASE, timeout=5.0) as client:
        try:
            async with asyncio.timeout(0.1):
                await fetch(client, "too-slow")
        except TimeoutError:
            print("1. deadline of 0.1s exceeded: TimeoutError, request cancelled")

        async def long_job():
            try:
                await asyncio.sleep(5)
            except asyncio.CancelledError:
                print("3. long_job saw CancelledError and cleaned up")
                raise

        task = asyncio.create_task(long_job())
        await asyncio.sleep(0.05)
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            print("2. awaiting the cancelled task raised CancelledError")

        try:
            await client.get("/item", params={"name": "slow"}, timeout=0.05)
        except httpx.TimeoutException as error:
            print(f"4. client timeout: {type(error).__name__}")

        print(f"5. task cancelled: {task.cancelled()}")


asyncio.run(main())

1. deadline of 0.1s exceeded: TimeoutError, request cancelled
3. long_job saw CancelledError and cleaned up
2. awaiting the cancelled task raised CancelledError
4. client timeout: ReadTimeout
5. task cancelled: True

A timeout is a cancellation with a timer attached. Inside the cancelled coroutine, that arrives as CancelledError, and it is raised at the next await point, which is why cleanup code belongs in a finally or an except asyncio.CancelledError that re-raises. Swallowing it without re-raising leaves the task in a state where the loop thinks it finished normally, and the caller waits forever.

There are two levels and you will want both. A per-request timeout on the client handles one slow endpoint; a deadline around a whole job handles the case where two hundred requests each finish inside their own timeout and the batch still exceeds your budget.

One client, many concurrent requests

Here is the comparison that matters for the service in this module. Eight requests to a server that sleeps 0.3 seconds each, one client reused for all of them:

import asyncio
import time

import httpx

BASE = "http://127.0.0.1:8801"
NAMES = [f"item-{n}" for n in range(1, 9)]


async def fetch(client, name):
    response = await client.get("/item", params={"name": name})
    response.raise_for_status()
    return response.json()["name"]


async def sequential(client):
    return [await fetch(client, name) for name in NAMES]


async def gathered(client):
    return await asyncio.gather(*(fetch(client, name) for name in NAMES))


async def grouped(client):
    async with asyncio.TaskGroup() as group:
        tasks = [group.create_task(fetch(client, name), name=name) for name in NAMES]
    return [task.result() for task in tasks]


async def main():
    async with httpx.AsyncClient(base_url=BASE, timeout=5.0) as client:
        started = time.perf_counter()
        first = await sequential(client)
        sequential_elapsed = time.perf_counter() - started

        started = time.perf_counter()
        second = await gathered(client)
        gathered_elapsed = time.perf_counter() - started

        started = time.perf_counter()
        third = await grouped(client)
        grouped_elapsed = time.perf_counter() - started

    print(f"await in a loop   {sequential_elapsed:5.2f}s")
    print(f"asyncio.gather    {gathered_elapsed:5.2f}s")
    print(f"asyncio.TaskGroup {grouped_elapsed:5.2f}s")
    print(f"all three agree: {first == second == third == NAMES}")


asyncio.run(main())

8 requests, server delay 0.3s each, one client
await in a loop    2.46s
asyncio.gather     0.31s
asyncio.TaskGroup  0.32s
all three agree: True

Two details in that script are doing real work. The client is created once and reused, which keeps the connection pool alive; a new AsyncClient per request costs a TCP handshake, and with TLS it costs far more. And the requests are unbounded, which is fine for eight and is not fine for eight thousand. A semaphore caps how many are in flight at once:

limit = asyncio.Semaphore(16)


async def bounded_fetch(client, name):
    async with limit:
        return await fetch(client, name)

The semaphore counts acquisitions, not connections, so it also bounds how many responses you hold in memory at the same time. Set it to something the server can tolerate and log the value at start-up, because the failure mode of an unbounded concurrency limit is a support ticket from whoever runs the other end.

import asyncio

import httpx

BASE = "http://127.0.0.1:8803"
PATHS = [f"/item?name=item-{n}" for n in range(1, 13)]
SLOW = {3, 7, 11}
results = {}


async def fetch(client, index, path):
    seconds = 30.0 if index in SLOW else 1.0
    try:
        response = await client.get(path, timeout=seconds)
    except httpx.TimeoutException:
        results[index] = "client-timeout"
        return
    results[index] = response.json()["name"]


async def main():
    async with httpx.AsyncClient(base_url=BASE) as client:
        try:
            async with asyncio.timeout(2.0):
                async with asyncio.TaskGroup() as group:
                    for index, path in enumerate(PATHS, start=1):
                        group.create_task(fetch(client, index, path), name=f"i{index}")
        except TimeoutError:
            print(f"batch deadline hit; {len(results)} of {len(PATHS)} finished")
    for index in range(1, 13):
        print(f"  {index:>2} {results.get(index, 'cancelled')}")


asyncio.run(main())

batch deadline hit; 9 of 12 finished
   1 item-1
   2 item-2
   3 cancelled
   4 item-4
   5 item-5
   6 item-6
   7 cancelled
   8 item-8
   9 item-9
  10 item-10
  11 item-11
  12 item-12