Asynchroner Code mit asyncio

Warten, ohne stillzustehen

Meistens rechnet ein Datenprogramm nicht, es wartet: auf die Antwort eines Preisservers, auf die Bestätigung einer Order, auf das Ende eines Downloads. Fragt es zehn Preise nacheinander ab, wartet es zehnmal hintereinander. Asynchroner Code lässt ein Programm eine Anfrage starten, sie beiseitelegen, solange sie wartet, und mit einer anderen weitermachen, alles in einem einzigen Thread. Das Python-Werkzeug dafür ist asyncio, gebaut aus zwei Schlüsselwörtern, async def und 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 und die Ereignisschleife

Eine mit async def definierte Funktion ist eine Coroutine-Funktion: Ihr Aufruf liefert ein Coroutine-Objekt und führt nichts vom Rumpf aus. Darin pausiert await etwas die Coroutine, bis etwas bereit ist, und gibt die Kontrolle an die Ereignisschleife zurück, die währenddessen andere Coroutinen ausführen kann. asyncio.run(coro) startet eine Schleife und führt eine Coroutine bis zum Ende aus. await gibt es nur innerhalb einer async def; es zu vergessen ist der klassische Fehler, denn der Aufruf liefert dann eine Coroutine statt der Antwort.

# 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

Fehler, Timeouts und von Hand fahren

Warten kann scheitern. Eine Anfrage, die nie antwortet, schneidet man mit asyncio.wait_for(coro, timeout) ab, was TimeoutError auslöst; ein await in try / except behandelt Fehler genau wie gewöhnlicher Code, ein Wiederholungsversuch ist also eine Schleife um einen erwarteten Aufruf. Unter der Haube ist eine Coroutine etwas, das man Schritt für Schritt vorantreiben kann: coro.send(None) führt sie bis zur nächsten Pause aus und löst beim Ende StopIteration mit dem Ergebnis aus. Eine Ereignisschleife ist der Code, der das für viele Coroutinen zugleich tut, und so führen auch die Tests dieses Labs Ihre aus.

# 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 definiert eine Coroutine-Funktion; der Aufruf erzeugt nur eine Coroutine, noch läuft nichts.
• await pausiert, bis das Ergebnis bereit ist, und lässt die Ereignisschleife andere ausführen; es gibt es nur in async def.
• asyncio.run startet die Schleife, gather überlappt Wartezeiten, wait_for ergänzt ein Zeitlimit; Fehler behandelt man mit gewöhnlichem try / except.

Als Nächstes: Streaming-Daten mit Websockets, einen Live-Feed von Ticks empfangen, während sie passieren.

Lab: Was bewertet wird

• Mit get_price auf einen Kurs warten25 Pkt.
• Mit get_prices eine Watchlist abrufen25 Pkt.
• Mit retry_price einen Timeout überstehen25 Pkt.
• Mit drive die Ereignisschleife sein25 Pkt.