Código asíncrono con asyncio

Esperar sin quedarse parado

La mayor parte del tiempo un programa de datos no calcula, espera: a que un servidor de precios responda, a que se confirme una orden, a que termine una descarga. Si pide diez precios uno por uno, espera diez veces seguidas. El código asíncrono permite que un programa inicie una petición, la deje de lado mientras espera y siga con otra, todo en un único hilo. La herramienta de Python para ello es asyncio, construida con dos palabras clave, async def y 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 y el bucle de eventos

Una función definida con async def es una función corrutina: llamarla devuelve un objeto corrutina y no ejecuta nada de su cuerpo. Dentro, await algo pausa la corrutina hasta que algo esté listo y devuelve el control al bucle de eventos, que puede ejecutar otras corrutinas mientras tanto. asyncio.run(coro) inicia un bucle y ejecuta una corrutina hasta el final. Solo se puede usar await dentro de una async def; olvidarlo es el error clásico, porque la llamada devuelve una corrutina en lugar de la respuesta.

# 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

Errores, tiempos de espera y hacerlo a mano

La espera puede fallar. Una petición que nunca responde debe cortarse con asyncio.wait_for(coro, timeout), que lanza TimeoutError; un await dentro de try / except gestiona los errores igual que el código normal, así que un reintento es un bucle alrededor de una llamada esperada. Por dentro, una corrutina es algo que se puede avanzar paso a paso: coro.send(None) la ejecuta hasta su próxima pausa y, al terminar, lanza StopIteration con el resultado. Un bucle de eventos es el código que hace eso sin parar con muchas corrutinas a la vez, y así es también como las pruebas de este laboratorio ejecutan las tuyas.

# 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 define una función corrutina; llamarla solo crea una corrutina, aún no se ejecuta nada.
• await pausa hasta que el resultado esté listo y deja que el bucle de eventos ejecute otras; solo funciona dentro de async def.
• asyncio.run inicia el bucle, gather solapa las esperas, wait_for añade un tiempo límite; los errores se gestionan con un try / except normal.

A continuación: datos en streaming con websockets, recibir un flujo de ticks en vivo según ocurren.

Lab: qué se califica

• Esperar una cotización con get_price25 pts
• Obtener una lista de seguimiento con get_prices25 pts
• Sobrevivir a un tiempo de espera con retry_price25 pts
• Ser el bucle de eventos con drive25 pts