async and await ⚡

The idea

A thread waiting on the network is a thread doing nothing, holding several megabytes of stack. async replaces that with a coroutine: a function that can pause itself at a marked point, hand control back, and be resumed later. Thousands of them fit in the memory one thread would use, and they all take turns on a single thread.

Your first coroutine

import asyncio


async def greet(name):
    """async def makes a coroutine function."""
    await asyncio.sleep(0.01)          # pause here, let others run
    return f"Hello, {name}"


async def main():
    result = await greet("Guybrush")
    print(result)


asyncio.run(main())

Hello, Guybrush

import asyncio


async def greet(name):
    return f"Hello, {name}"


coro = greet("Elaine")
print(type(coro))
print(asyncio.run(coro))

<class 'coroutine'>
Hello, Elaine

The payoff: doing many things at once

import asyncio, time


async def fetch(page):
    await asyncio.sleep(0.1)           # pretend network delay
    return f"page {page}"


async def one_at_a_time():
    return [await fetch(n) for n in range(5)]


async def all_at_once():
    return await asyncio.gather(*(fetch(n) for n in range(5)))


start = time.perf_counter()
asyncio.run(one_at_a_time())
sequential = time.perf_counter() - start

start = time.perf_counter()
results = asyncio.run(all_at_once())
concurrent = time.perf_counter() - start

print(results)
print(f"sequential about 0.5s: {0.4 < sequential < 0.9}")
print(f"concurrent about 0.1s: {concurrent < 0.3}")

['page 0', 'page 1', 'page 2', 'page 3', 'page 4']
sequential about 0.5s: True
concurrent about 0.1s: True

Tasks: starting work in the background

import asyncio


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


async def main():
    slow = asyncio.create_task(work("slow", 0.2))     # starts immediately
    fast = asyncio.create_task(work("fast", 0.05))

    print("both are now running")
    results = [await fast, await slow]
    return results


print(asyncio.run(main()))

both are now running
  fast done
  slow done
['fast', 'slow']

TaskGroup: the modern, safer way

import asyncio


async def fetch(page):
    await asyncio.sleep(0.01)
    if page == 3:
        raise ValueError("page 3 is missing")
    return f"page {page}"


async def main():
    failures = []
    try:
        async with asyncio.TaskGroup() as group:
            tasks = [group.create_task(fetch(n)) for n in range(5)]
    except* ValueError as errors:
        failures.extend(errors.exceptions)      # note: no `return` in here

    if failures:
        print(f"caught {len(failures)} failure(s):", failures[0])
        return "aborted"
    return [t.result() for t in tasks]


print(asyncio.run(main()))

caught 1 failure(s): page 3 is missing
aborted

TaskGroup (Python 3.11+) guarantees that every task finishes or is cancelled before the block exits, and it collects failures into an ExceptionGroup caught with except*. Before this existed it was easy to leave orphaned tasks running silently after an error. Prefer it to bare gather in new code.

One rule that catches people: return, break and continue are not allowed inside an except* block, because an exception group can trigger several handlers and Python refuses to guess which return wins. Collect what you need into a variable, then act on it after the block, which is what the example above does.

Timeouts, which you will always need

import asyncio


async def slow():
    await asyncio.sleep(10)
    return "eventually"


async def main():
    try:
        async with asyncio.timeout(0.05):
            return await slow()
    except TimeoutError:
        return "gave up waiting"


print(asyncio.run(main()))

gave up waiting

The rules of the road

• async is contagious. To await something you must be in an async def, whose caller must await it too, all the way up to asyncio.run. People call this "function colouring", and it is the main complaint about async in every language that has it.
• Never block inside a coroutine. One time.sleep(5) or one ordinary requests.get freezes the entire event loop, and every other task with it. Use the async equivalents.
• Your libraries must cooperate. requests is blocking; you want httpx or aiohttp. Databases need async drivers too.

import asyncio, time


async def blocking_mistake():
    time.sleep(0.1)          # WRONG: freezes everything
    return "blocked"


async def correct():
    await asyncio.sleep(0.1)  # right: yields control
    return "yielded"


async def escape_hatch():
    """When you must call blocking code, push it to a thread."""
    return await asyncio.to_thread(time.sleep, 0.01) or "ran in a thread"


async def main():
    return [await correct(), await escape_hatch()]


print(asyncio.run(main()))

['yielded', 'ran in a thread']

Async iteration

import asyncio


async def stream_pages(count):
    """An async generator: yields values as they become available."""
    for n in range(count):
        await asyncio.sleep(0.01)
        yield f"page {n}"


async def main():
    async for page in stream_pages(3):
        print("received", page)

    results = [p async for p in stream_pages(2)]
    return results


print(asyncio.run(main()))

received page 0
received page 1
received page 2
['page 0', 'page 1']

This is exactly how you will consume a streaming response from a language model in Level 6: tokens arrive one at a time, and async for processes each as it lands rather than waiting for the whole reply.

Threads or async?

import asyncio


async def check(site):
    await asyncio.sleep(0.1)
    return f"{site}: ok"


async def main():
    results = []
    for site in ["a.com", "b.com", "c.com", "d.com"]:
        results.append(await check(site))
    return results

import asyncio


async def check(site):
    await asyncio.sleep(0.1)
    return f"{site}: ok"


async def main():
    sites = ["a.com", "b.com", "c.com", "d.com"]
    async with asyncio.TaskGroup() as group:
        tasks = [group.create_task(check(s)) for s in sites]
    return [t.result() for t in tasks]


for line in asyncio.run(main()):
    print(line)

a.com: ok
b.com: ok
c.com: ok
d.com: ok

import asyncio


async def fetch_slowly(delay, value):
    await asyncio.sleep(delay)
    return value


async def with_fallback(coro, seconds, default):
    """Await coro, or return default if it takes longer than seconds."""
    try:
        async with asyncio.timeout(seconds):
            return await coro
    except TimeoutError:
        return default


async def main():
    quick = await with_fallback(fetch_slowly(0.01, "live data"), 0.1, "cached")
    slow = await with_fallback(fetch_slowly(1.0, "live data"), 0.05, "cached")
    return quick, slow


print(asyncio.run(main()))

('live data', 'cached')

import asyncio, time


async def process(item):
    time.sleep(0.1)          # <- here
    return item * 2


async def main():
    return await asyncio.gather(*(process(n) for n in range(10)))

import asyncio


async def process(item):
    await asyncio.sleep(0.1)          # yields control properly
    return item * 2


async def main():
    return await asyncio.gather(*(process(n) for n in range(10)))


print(asyncio.run(main()))

[0, 2, 4, 6, 8, 10, 12, 14, 16, 18]