# Async and Await

> Source: https://learn-python.com/async-await/
> Part of Learn Python, free to read.

Async is for **waiting on many things at once** — network calls, database queries, file I/O. It does not make computation faster; it stops you idling while the network responds.

```python
import asyncio

async def fetch(name: str, delay: float) -> str:
    await asyncio.sleep(delay)         # yields control while waiting
    return f"{name} done"

async def main() -> None:
    result = await fetch("first", 1.0)
    print(result)

asyncio.run(main())
```

`async def` creates a coroutine. Calling it does nothing on its own — it returns a coroutine object that must be awaited or scheduled:

```python
fetch("a", 1)            # RuntimeWarning: coroutine was never awaited
await fetch("a", 1)      # runs it
```

## Concurrency is the point

```python
async def main() -> None:
    a = await fetch("first", 1.0)      # 1 second
    b = await fetch("second", 1.0)     # then another second
    # total: 2 seconds — this is just slow synchronous code
```

Sequential `await`s buy you nothing. Run them together:

```python
async def main() -> None:
    a, b = await asyncio.gather(
        fetch("first", 1.0),
        fetch("second", 1.0),
    )
    # total: 1 second
```

**Consecutive `await` lines with no data dependency between them are the signal** that something should be a `gather`.

## TaskGroup

The modern replacement for `gather`, and the better default:

```python
async def main() -> None:
    async with asyncio.TaskGroup() as tg:
        first = tg.create_task(fetch("first", 1.0))
        second = tg.create_task(fetch("second", 1.0))
    # both are complete here
    print(first.result(), second.result())
```

The advantage over `gather` is failure handling: if one task raises, the others are **cancelled** and the errors are collected into an `ExceptionGroup`. With `gather`, a failure leaves the siblings running in the background, still consuming resources.

```python
try:
    async with asyncio.TaskGroup() as tg:
        ...
except* ValueError as eg:              # except* handles ExceptionGroup
    for err in eg.exceptions:
        print("bad value:", err)
```

## The mistake that matters

:::danger A blocking call inside `async def` freezes everything
```python
import requests, time

async def fetch_user(uid: str):
    r = requests.get(f"{API}/users/{uid}")   # BLOCKS the entire event loop
    time.sleep(0.2)                          # so does this
    return r.json()
```
There is one event loop and one thread. A blocking call means every other coroutine stops — so your "concurrent" server handles one request at a time, and the symptom is latency under load rather than an error.

This is the single most damaging bug in async Python and it is very common in generated code, because `requests` and `time.sleep` are far more represented in the training data than their async equivalents.
:::

| Blocking | Async |
|---|---|
| `requests.get()` | `httpx.AsyncClient().get()`, `aiohttp` |
| `time.sleep()` | `await asyncio.sleep()` |
| `open()` / `read()` | `aiofiles`, or `asyncio.to_thread` |
| a synchronous DB driver | `asyncpg`, `aiosqlite`, an async SQLAlchemy engine |
| CPU-heavy work | `asyncio.to_thread` (I/O) or a process pool (CPU) |

```python
result = await asyncio.to_thread(expensive_sync_function, arg)
```

`to_thread` runs blocking code in a worker thread so the loop keeps running. It is the escape hatch for a library with no async version.

Catch it mechanically: turn on `ruff`'s `ASYNC` rules, and run tests with `PYTHONASYNCIODEBUG=1`, which warns when a coroutine blocks for more than 100ms. See [the failure-mode catalogue](/review/failure-modes/).

## Bounding concurrency

`gather` over a large list starts everything at once — which means a rate-limited API returns 429s, or you exhaust file descriptors:

```python
async def fetch_all(ids: list[str]) -> list[str]:
    sem = asyncio.Semaphore(10)          # at most 10 in flight

    async def one(uid: str) -> str:
        async with sem:
            return await fetch(uid, 0.1)

    return await asyncio.gather(*(one(i) for i in ids))
```

When the list size comes from data rather than from your source code, bound it.

## Timeouts and cancellation

```python
try:
    async with asyncio.timeout(5.0):
        data = await slow_operation()
except TimeoutError:
    print("gave up after 5 seconds")
```

Cancellation is cooperative: `asyncio` raises `CancelledError` at the next `await` point. Clean up with `try/finally`, and **never swallow it**:

```python
async def worker() -> None:
    try:
        while True:
            await do_work()
    except asyncio.CancelledError:
        await flush()      # clean up
        raise              # ALWAYS re-raise — swallowing breaks shutdown
    finally:
        await close()
```

## Async iterators and context managers

```python
async with httpx.AsyncClient() as client:      # __aenter__ / __aexit__
    r = await client.get(url)

async for row in cursor:                        # __aiter__ / __anext__
    process(row)

async def paginate(url: str):
    while url:
        page = await fetch_page(url)
        for item in page["items"]:
            yield item                          # an async generator
        url = page.get("next")

async for item in paginate("/api/items"):
    print(item)
```

## When not to use async

Async adds real complexity — a coloured-function split where async code can only be called from async code, harder debugging, and a whole category of new bugs.

**Do not use it for:** CPU-bound work (use `multiprocessing`), scripts making a handful of sequential calls, or code where the concurrency would be one or two operations. Threads are simpler and often sufficient.

**Do use it for:** servers handling many simultaneous connections, clients making hundreds of requests, and anything where you would otherwise be waiting on I/O most of the time.

## Exercise

```python
import asyncio

async def fetch_price(sku: str) -> int:
    await asyncio.sleep(0.1)
    return len(sku) * 100

async def main() -> None:
    skus = ["A1", "B22", "C333", "D4444"]
    # 1. Fetch all prices CONCURRENTLY and print the total.
    # 2. Limit concurrency to 2 with a Semaphore.
    # 3. Wrap the whole thing in a 1-second timeout.
    print(skus)

asyncio.run(main())
```

## Common questions

### Does async make my code faster?

Only for I/O-bound work, and only when operations actually overlap. It removes waiting, not computation — a CPU-heavy function is exactly as slow inside a coroutine, and it blocks everything else while it runs.

### Why is my async code no faster than the sync version?

The two usual causes: you are awaiting sequentially instead of using `gather` or a `TaskGroup`, or something in the path is a blocking call that freezes the loop. Check for `requests`, `time.sleep` or a synchronous database driver first.

### `gather` or `TaskGroup`?

`TaskGroup` for new code. When one task fails it cancels the siblings and reports errors as a group, whereas `gather` leaves the others running in the background. Use `gather` when you genuinely want independent results and `return_exceptions=True`.