# Async Python

> Run many I/O operations concurrently with asyncio: coroutines, await, TaskGroup, timeouts, limiting concurrency, avoiding blocking calls, and when not to use async.

Python roadmap · Building things · 18 min read · https://levelcode.org/roadmaps/python/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:

```python
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:

```python
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+):

```python
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())
```

```text
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:

```python
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:

```python
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**:

```python
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 `await`s, **nothing else can run**. A blocking call inside a coroutine freezes every other task:

```python
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:

```python
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")
```

> [!TIP]
> Run with `PYTHONASYNCIODEBUG=1` (or `asyncio.run(main(), debug=True)`) during development. asyncio then logs a warning whenever a single step blocks the loop for longer than 100 ms, pointing straight at the blocking call.

## 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

| Choose async when | Stay synchronous when |
|---|---|
| Many concurrent network calls: scrapers, API fan-out, LLM calls | A script that makes a few calls in sequence |
| Servers handling many slow, concurrent connections (WebSockets, streaming) | CPU-bound work: use processes instead |
| The framework is async already (FastAPI, aiohttp) | Your key libraries have no async versions |

Async adds real complexity: two kinds of functions, a new class of bugs (blocking the loop, forgotten `await`s, 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`.

**Solution**

```python
# 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}")
```

```python
# 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
```

Each task catches its own expected errors and turns them into values, which is why one failure doesn't cancel the group: only unexpected exceptions (bugs) escape and cancel the rest, which is what you want for bugs. Accepting an optional `client` makes the function testable; when none is passed, `check` creates one and closes it itself. With 20 URLs at 0.1 s each and a limit of 10, the requests run in two waves, about 0.2 s in total.

## 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.

## Quiz

1. Which workload benefits most from asyncio?

   A. Resizing 10,000 images
   B. Calling a slow API 500 times
   C. Computing prime numbers
   D. Sorting a large list in memory

   **Answer: B.** API calls spend nearly all their time waiting on the network, which async overlaps. The others are CPU-bound and need processes, not an event loop.

2. Two coroutines each await asyncio.sleep(2). How long does a = await one(); b = await two() take?

   A. 2 seconds
   B. 0 seconds
   C. 4 seconds
   D. It depends on the number of CPU cores

   **Answer: C.** Awaiting one after the other is sequential. To overlap them, create tasks, for example in a TaskGroup, which would take about 2 seconds.

3. An async web server freezes for every user while one request runs. What's the most likely cause?

   A. A blocking call such as time.sleep or requests.get inside a coroutine
   B. Too many await statements
   C. Using TaskGroup
   D. The semaphore limit is too high

   **Answer: A.** All coroutines share one thread. A blocking call stops the event loop, so no other request can make progress until it returns.

4. In an asyncio.TaskGroup, one task raises an unexpected exception. What happens to the others?

   A. They keep running in the background
   B. They are paused until you resume them
   C. Nothing; the exception is ignored
   D. They are cancelled, and the errors are raised together as an ExceptionGroup

   **Answer: D.** TaskGroup provides structured concurrency: a failure cancels the remaining tasks, waits for them, and raises everything together.