Async Python

A program that fetches 100 web pages one after another spends almost all its time waiting: each request takes 200 ms, of which a few microseconds are actual work. Asynchronous programming lets one thread start many such operations and work on whichever is ready, so 100 requests take roughly as long as the slowest one, not the sum of all of them. This lesson explains how asyncio does that, how to write and run coroutines, run them concurrently and safely with TaskGroup, add timeouts and limits, and the mistakes that quietly make async code slow.

#Concurrency for waiting, not for computing

There are two kinds of slow:

• I/O-bound work waits on something outside the CPU: the network, a database, a disk. Most web backends, scrapers, API clients and LLM applications are I/O-bound.
• CPU-bound work keeps the processor busy: resizing images, parsing huge files, numerical computation.

Async helps the first kind only. While one operation waits for the network, the program runs another. It does not make computation faster, because everything still runs on one thread, one step at a time. For CPU-bound work, use multiple processes (concurrent.futures.ProcessPoolExecutor) or libraries that do the work in native code.

#Coroutines and await

An async def function is a coroutine function. Calling it doesn't run it; it creates a coroutine object that runs when awaited:

import asyncio


async def make_coffee(name: str, seconds: float) -> str:
    print(f"start {name}")
    await asyncio.sleep(seconds)       # wait without blocking anything else
    print(f"done {name}")
    return f"{name} coffee"


async def main() -> None:
    result = await make_coffee("espresso", 1)
    print(result)


asyncio.run(main())

asyncio.run starts the event loop, the scheduler at the heart of asyncio, runs main() to completion, and shuts the loop down. Call it once, at the entry point of your program.

await means "pause this coroutine until that result is ready, and let the event loop run other things meanwhile". You can only use await inside async def. That's why async tends to spread upwards through a codebase: a function that awaits must itself be async.

#Running things concurrently

Awaiting coroutines one after another is still sequential:

async def main() -> None:
    a = await make_coffee("espresso", 1)
    b = await make_coffee("latte", 2)     # starts only after the espresso is done: 3 s total

To overlap them, turn them into tasks, which the event loop runs concurrently. The modern, safe way is asyncio.TaskGroup (Python 3.11+):

import time


async def main() -> None:
    start = time.perf_counter()
    async with asyncio.TaskGroup() as tg:
        espresso = tg.create_task(make_coffee("espresso", 1))
        latte = tg.create_task(make_coffee("latte", 2))
    # the async with block waits for every task in the group
    print(espresso.result(), latte.result())
    print(f"took {time.perf_counter() - start:.1f}s")    # took 2.0s, not 3.0s


asyncio.run(main())

start espresso
start latte
done espresso
done latte
espresso coffee latte coffee
took 2.0s

Here's what happens. Both tasks start; each reaches await asyncio.sleep(...) and yields control; the event loop sleeps until the first timer fires, resumes the espresso, and later the latte.

TaskGroup also makes failure handling correct. If any task raises, the group cancels the other tasks, waits for them to finish cancelling, and raises the errors together as an ExceptionGroup, which you handle with except*. No task is left running unnoticed in the background.

asyncio.gather(*coros) is the older way to do the same and returns results in order. It's still common in existing code, but by default it doesn't cancel the remaining tasks when one fails, so prefer TaskGroup in new code.

#An async HTTP client

Concurrency pays off with real I/O. httpx has an async client with the same API as the sync one:

import asyncio

import httpx

URLS = [f"https://httpbin.org/delay/1?n={i}" for i in range(10)]


async def fetch_status(client: httpx.AsyncClient, url: str) -> int:
    response = await client.get(url)
    return response.status_code


async def main() -> None:
    async with httpx.AsyncClient(timeout=10) as client:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch_status(client, url)) for url in URLS]
    print([t.result() for t in tasks])


asyncio.run(main())    # about 1 second for all 10, instead of about 10

The client is shared by all tasks: it pools connections and is safe for concurrent use within one event loop.

#Timeouts

Wrap anything that could hang in asyncio.timeout (Python 3.11+), which cancels the work if the deadline passes:

async def fetch_with_deadline(client: httpx.AsyncClient, url: str) -> int | None:
    try:
        async with asyncio.timeout(2):
            response = await client.get(url)
            return response.status_code
    except TimeoutError:
        return None

The timeout covers everything inside the block, however many awaits it contains, which makes it a true deadline rather than a per-call limit. asyncio.wait_for(coro, timeout) is the older equivalent for a single awaitable.

#Limiting concurrency

Starting 10,000 tasks at once will open 10,000 connections, trip the API's rate limit, or exhaust file handles. Bound the number in flight with a semaphore:

async def fetch_all(urls: list[str], limit: int = 20) -> list[int | None]:
    semaphore = asyncio.Semaphore(limit)

    async def fetch_one(client: httpx.AsyncClient, url: str) -> int | None:
        async with semaphore:                 # at most `limit` tasks get past here at once
            return await fetch_with_deadline(client, url)

    async with httpx.AsyncClient() as client:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch_one(client, url)) for url in urls]
    return [t.result() for t in tasks]

All tasks are created, but only 20 run their request at any time; the rest wait at async with semaphore. Pick the limit from what the other side tolerates, not from what your machine can do.

#The cardinal sin: blocking the event loop

Everything in an asyncio program shares one thread. While a coroutine is running between two awaits, nothing else can run. A blocking call inside a coroutine freezes every other task:

import time
import requests


async def bad() -> None:
    time.sleep(2)                     # blocks the whole program for 2 s
    requests.get("https://example.com")   # a synchronous library: also blocks everything

The symptoms are confusing: your "concurrent" program runs sequentially, timeouts fire late, and a server stops responding to all users while one request does something slow. Rules:

• Use await asyncio.sleep(), never time.sleep().
• Use async libraries for I/O: httpx.AsyncClient (not requests), asyncpg or psycopg's async API for PostgreSQL, aiofiles or threads for files.
• When you must call blocking code (a sync-only SDK, heavy parsing), push it to a thread with asyncio.to_thread, which keeps the event loop free:

import hashlib


def hash_file(path: str) -> str:            # ordinary blocking function
    with open(path, "rb") as f:
        return hashlib.sha256(f.read()).hexdigest()


async def main() -> None:
    digest = await asyncio.to_thread(hash_file, "big.iso")

#Cancellation

When a task is cancelled (by a TaskGroup, a timeout, or task.cancel()), asyncio raises CancelledError inside it at its current await. Let it propagate. Use try/finally or async with for cleanup, and never swallow CancelledError with a broad except. Since Python 3.8 it inherits from BaseException, so except Exception doesn't catch it by accident, which is exactly why you shouldn't write except BaseException in async code without re-raising.

#When to use async, and when not

Async adds real complexity: two kinds of functions, a new class of bugs (blocking the loop, forgotten awaits, which produce "coroutine was never awaited" warnings), and less mature tooling for some libraries. Use it where its benefit, many simultaneous waits, is what you need.

#Practice

Write checker.py, a link checker:

1. check(urls, limit=10, timeout=5) fetches every URL concurrently with an httpx.AsyncClient, at most limit at a time, each with a timeout-second deadline.
2. It returns a dict mapping each URL to its status code, or to a short error string ("timeout", "connection error").
3. One failing URL must not stop the others.
4. Test it with httpx.MockTransport (it works with AsyncClient too) using URLs that return 200, 404, raise httpx.ConnectError and are slow, and check that 20 URLs that each take 0.1 s complete in well under 2 seconds with limit=10.

# checker.py
import asyncio

import httpx


async def _check_one(client: httpx.AsyncClient, sem: asyncio.Semaphore, url: str, timeout: float) -> int | str:
    async with sem:
        try:
            async with asyncio.timeout(timeout):
                response = await client.get(url)
            return response.status_code
        except TimeoutError:
            return "timeout"
        except httpx.TransportError:
            return "connection error"


async def check(
    urls: list[str], limit: int = 10, timeout: float = 5, client: httpx.AsyncClient | None = None
) -> dict[str, int | str]:
    sem = asyncio.Semaphore(limit)
    owns_client = client is None
    client = client or httpx.AsyncClient(follow_redirects=True)
    try:
        async with asyncio.TaskGroup() as tg:
            tasks = {url: tg.create_task(_check_one(client, sem, url, timeout)) for url in urls}
    finally:
        if owns_client:
            await client.aclose()
    return {url: task.result() for url, task in tasks.items()}


if __name__ == "__main__":
    import sys
    for url, result in asyncio.run(check(sys.argv[1:])).items():
        print(f"{result!s:>16}  {url}")

# test_checker.py
import asyncio
import time

import httpx

from checker import check


async def handler(request: httpx.Request) -> httpx.Response:
    match request.url.path:
        case "/ok":
            return httpx.Response(200)
        case "/missing":
            return httpx.Response(404)
        case "/down":
            raise httpx.ConnectError("refused", request=request)
        case "/slow":
            await asyncio.sleep(1)
            return httpx.Response(200)
        case _:
            await asyncio.sleep(0.1)
            return httpx.Response(200)


def client() -> httpx.AsyncClient:
    return httpx.AsyncClient(transport=httpx.MockTransport(handler), base_url="https://test")


def test_mixed_results():
    urls = ["https://test/ok", "https://test/missing", "https://test/down", "https://test/slow"]
    result = asyncio.run(check(urls, timeout=0.5, client=client()))
    assert result == {
        "https://test/ok": 200,
        "https://test/missing": 404,
        "https://test/down": "connection error",
        "https://test/slow": "timeout",
    }


def test_runs_concurrently():
    urls = [f"https://test/page{i}" for i in range(20)]
    start = time.perf_counter()
    result = asyncio.run(check(urls, limit=10, client=client()))
    elapsed = time.perf_counter() - start
    assert all(status == 200 for status in result.values())
    assert elapsed < 0.5        # two waves of 0.1 s, not twenty

#Key takeaways

• Async speeds up I/O-bound work by overlapping waits on one thread; it doesn't speed up computation.
• async def creates coroutines, await pauses one until a result is ready, and asyncio.run starts the event loop once at the entry point.
• Run work concurrently with asyncio.TaskGroup, which cancels siblings and raises an ExceptionGroup on failure.
• Bound waiting with asyncio.timeout and bound load with asyncio.Semaphore.
• Never block the loop: use async libraries, asyncio.sleep, and asyncio.to_thread for unavoidable blocking calls.