Asynchronous code with asyncio

Waiting without standing still

Most of the time a data program is not computing, it is waiting: for a price server to answer, for an order to be acknowledged, for a file to download. If it asks for ten prices one by one, it waits ten times in a row. Asynchronous code lets one program start a request, set it aside while it waits, and get on with another, all in a single thread. The Python tool for it is asyncio, built from two keywords, async def and await.

# A coroutine: calling it does nothing until it is awaited
import asyncio

async def get_price(symbol):
    await asyncio.sleep(0.01)                  # stands in for waiting on a network answer
    return {"AAPL": 184.25, "MSFT": 410.5}[symbol]

coro = get_price("AAPL")
print(type(coro).__name__)                     # coroutine  nothing has run yet
print(asyncio.run(coro))                       # 184.25     the event loop runs it

async def, await and the event loop

A function defined with async def is a coroutine function: calling it returns a coroutine object and runs none of its body. Inside it, await something pauses the coroutine until something is ready and hands control back to the event loop, which can run other coroutines meanwhile. asyncio.run(coro) starts a loop and runs one coroutine to the end. You can only use await inside an async def; forgetting it is the classic mistake, because the call then returns a coroutine instead of the answer.

# Awaiting ten prices one after another waits ten times; gather overlaps the waits
import asyncio
import time

async def get_price(symbol):
    await asyncio.sleep(0.05)
    return len(symbol) * 10.0

async def one_by_one(symbols):
    return [await get_price(s) for s in symbols]

async def together(symbols):
    return await asyncio.gather(*(get_price(s) for s in symbols))

symbols = ["AAPL", "MSFT", "NVDA", "TSLA", "AMZN", "META", "GOOG", "NFLX", "ORCL", "INTC"]
t0 = time.perf_counter()
asyncio.run(one_by_one(symbols))
slow = time.perf_counter() - t0
t0 = time.perf_counter()
asyncio.run(together(symbols))
fast = time.perf_counter() - t0
print(f"one by one: {slow:.2f} s, gathered: {fast:.2f} s")      # about 0.50 s against 0.05 s
print(slow > 4 * fast)                                           # True

Errors, timeouts and doing it by hand

Waiting can fail. A request that never answers should be cut off with asyncio.wait_for(coro, timeout), which raises TimeoutError; an await inside try / except handles errors exactly like ordinary code, so a retry is a loop around an awaited call. Under the hood a coroutine is just something you can step: coro.send(None) runs it until its next pause and, when it finishes, raises StopIteration carrying the result. An event loop is the code that keeps doing that for many coroutines at once, which is also how this lab's tests run yours.

# Stepping a coroutine by hand: this is what an event loop does for many of them
class Pause:
    def __await__(self):
        yield

async def job():
    await Pause()
    await Pause()
    return 184.25

coro = job()
steps = 0
while True:
    try:
        coro.send(None)
    except StopIteration as stop:
        print("finished with", stop.value)                  # finished with 184.25
        break
    steps += 1
print("paused", steps, "times")                              # paused 2 times

# A timeout turns a slow answer into an error you can handle, then retry
import asyncio

async def slow_feed():
    await asyncio.sleep(1.0)
    return 184.25

async def main():
    for attempt in range(1, 4):
        try:
            return await asyncio.wait_for(slow_feed(), timeout=0.01)
        except TimeoutError:
            print(f"attempt {attempt} timed out")
    return None

print(asyncio.run(main()))                                   # three timeouts, then None

• async def defines a coroutine function; calling it only creates a coroutine, nothing runs yet.
• await pauses until the result is ready and lets the event loop run others; it works only inside async def.
• asyncio.run starts the loop, gather overlaps waits, wait_for adds a timeout; errors are handled with ordinary try / except.

Next up: streaming data with websockets, receiving a live feed of ticks as they happen.

Lab · what's graded

• Wait for one quote with get_price25 pts
• Fetch a watchlist with get_prices25 pts
• Survive a timeout with retry_price25 pts
• Be the event loop with drive25 pts