1.16 · Asincronía con asyncio¶

Objetivos. Al terminar este capítulo podrás: diferenciar concurrencia de paralelismo; definir coroutines con async def; esperar operaciones con await; ejecutar varias tareas con asyncio.gather; y evitar bloquear el bucle de eventos.

Evidencia de logro. Escribirás un cliente simulado que consulta varias fuentes concurrentemente, conserva el orden de los resultados y trata un tiempo agotado sin depender de una red real.

Contexto y motivación¶

Una aplicación de IA suele esperar a servicios HTTP, discos o colas. Mientras una operación espera, el programa puede atender otra tarea. La asincronía organiza esa espera en un único hilo mediante un bucle de eventos.

• Concurrencia: varias tareas avanzan intercaladas durante sus esperas.
• Paralelismo: varias operaciones se ejecutan realmente al mismo tiempo, normalmente en varios núcleos o procesos.

asyncio es especialmente útil para muchas operaciones de entrada/salida. No hace que un cálculo pesado de Python sea más rápido por sí solo.

Vocabulario¶

Prerrequisitos¶

1.5 · Bucles, 1.8 · Funciones, 1.11 · Excepciones y logging y 1.15 · Librerías externas (esta última, solo necesaria para la ampliación del cliente HTTP).

1. Una coroutine no se ejecuta al definirla¶

async def crea una función que devuelve una coroutine. Para ejecutarla desde un script se usa asyncio.run:

import asyncio


async def saludar():
    await asyncio.sleep(0)
    return "Hola desde una coroutine"


print(asyncio.run(saludar()))

Salida esperada:

Hola desde una coroutine

await asyncio.sleep(0) cede el control al bucle. En una operación real sería la espera de una respuesta HTTP, una cola o un fichero asíncrono.

2. Esperar varias operaciones¶

Si esperas una operación después de otra, las esperas se suman. Con asyncio.gather pueden avanzar juntas. Compruébalo con tres esperas simuladas de 0,1 segundos cada una:

import asyncio
import time


async def esperar(segundos):
    await asyncio.sleep(segundos)      # simula esperar a un servidor
    return segundos


async def una_tras_otra():
    for _ in range(3):
        await esperar(0.1)


async def a_la_vez():
    await asyncio.gather(esperar(0.1), esperar(0.1), esperar(0.1))


inicio = time.perf_counter()
asyncio.run(una_tras_otra())
print("Una tras otra:", round(time.perf_counter() - inicio, 1), "s")

inicio = time.perf_counter()
asyncio.run(a_la_vez())
print("A la vez:", round(time.perf_counter() - inicio, 1), "s")

Salida esperada:

Una tras otra: 0.3 s
A la vez: 0.1 s

time.perf_counter() es un cronómetro: la diferencia entre dos lecturas es el tiempo transcurrido, que redondeamos a una décima porque varía unos milisegundos entre ejecuciones. Una tras otra, las tres esperas se suman (0,3 s); con gather, mientras una espera, las otras también esperan, y el total es aproximadamente el de la más lenta (0,1 s). Esa es toda la ventaja de la asincronía: aprovechar las esperas.

Para ver el orden en que avanzan, este ejemplo imprime al empezar y al terminar cada consulta (con sleep(0), que solo cede el turno sin esperar):

import asyncio


async def consultar(nombre):
    print("inicio", nombre)
    await asyncio.sleep(0)
    print("fin", nombre)
    return f"{nombre}: disponible"


async def consultar_todas():
    return await asyncio.gather(
        consultar("catálogo"),
        consultar("memoria"),
        consultar("modelo"),
    )


print(asyncio.run(consultar_todas()))

Salida esperada:

inicio catálogo
inicio memoria
inicio modelo
fin catálogo
fin memoria
fin modelo
['catálogo: disponible', 'memoria: disponible', 'modelo: disponible']

Las tareas avanzan hasta await, ceden el control y continúan después. Aunque el orden de impresión puede depender de dónde cedan las coroutines, gather devuelve los resultados en el orden de las operaciones que recibió.

Ampliación: crear y cancelar tareas¶

create_task programa una coroutine para que avance mientras hacemos otra cosa:

import asyncio


async def trabajo(nombre):
    try:
        await asyncio.sleep(0)
        return f"terminado: {nombre}"
    except asyncio.CancelledError:
        print("cancelado:", nombre)
        raise


async def main():
    tarea = asyncio.create_task(trabajo("consulta"))
    await asyncio.sleep(0)
    print(tarea.cancel())
    try:
        await tarea
    except asyncio.CancelledError:
        print("cancelación confirmada")


asyncio.run(main())

Salida esperada:

True
cancelado: consulta
cancelación confirmada

Cancelar no es lo mismo que ignorar el resultado: la coroutine recibe CancelledError y debe liberar recursos en un finally si los ha adquirido.

3. Timeouts y excepciones¶

Una espera sin límite puede dejar una aplicación ocupada indefinidamente. En Python 3.11 o posterior, asyncio.timeout expresa el límite:

import asyncio


async def esperar_servidor():
    await asyncio.sleep(0.02)
    return "respuesta"


async def main():
    try:
        async with asyncio.timeout(0.001):
            return await esperar_servidor()
    except TimeoutError:
        return "tiempo agotado"


print(asyncio.run(main()))

Salida esperada:

tiempo agotado

El timeout cancela la espera dentro del contexto. La capa que conoce el caso de uso decide si devuelve una respuesta alternativa, reintenta o informa del error. No conviertas todo timeout en una respuesta válida sin registrar la pérdida de información.

4. El error más común: bloquear el bucle¶

Este código bloquea todas las demás tareas durante la espera:

# Incorrecto dentro de una coroutine:
# time.sleep(2)

Usa una operación asíncrona (await asyncio.sleep(2)) si la librería la ofrece. Si solo existe una función síncrona y no puedes cambiarla, puedes moverla a un hilo:

import asyncio
import time


def lectura_síncrona():
    time.sleep(0)
    return "lectura terminada"


async def main():
    resultado = await asyncio.to_thread(lectura_síncrona)
    return resultado


print(asyncio.run(main()))

Salida esperada:

lectura terminada

to_thread evita bloquear el bucle durante una función de entrada/salida síncrona. No es una solución automática para cálculos intensivos: para CPU pesada se estudian procesos, vectorización u otras estrategias.

Ampliación: cliente HTTP asíncrono¶

En una aplicación real, httpx2.AsyncClient ofrece métodos asíncronos. Para practicar sin Internet levantamos un servidor HTTP local:

import asyncio
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from threading import Thread

import httpx2


class Manejador(BaseHTTPRequestHandler):
    def do_GET(self):
        contenido = b'{"estado":"ok"}'
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(contenido)))
        self.end_headers()
        self.wfile.write(contenido)

    def log_message(self, formato, *argumentos):
        pass


async def consultar():
    servidor = ThreadingHTTPServer(("127.0.0.1", 0), Manejador)
    hilo = Thread(target=servidor.serve_forever, daemon=True)
    hilo.start()
    try:
        async with httpx2.AsyncClient() as cliente:
            respuesta = await cliente.get(
                f"http://127.0.0.1:{servidor.server_port}/estado",
                timeout=5.0,
            )
            respuesta.raise_for_status()
            return respuesta.json()
    finally:
        servidor.shutdown()
        servidor.server_close()


print(asyncio.run(consultar()))

Salida esperada:

{'estado': 'ok'}

El cliente se abre y se cierra con async with. En un servicio web de larga vida suele crearse un cliente compartido y cerrarse durante el apagado de la aplicación; no abras uno nuevo para cada línea de un bucle sin motivo.

Aplicación práctica: consultar fuentes de un asistente¶

import asyncio


async def fuente(nombre, disponible=True):
    await asyncio.sleep(0)
    if not disponible:
        raise ConnectionError(f"{nombre} no responde")
    return {"fuente": nombre, "estado": "ok"}


async def consultar_fuentes():
    resultados = await asyncio.gather(
        fuente("notas"),
        fuente("calendario", disponible=False),
        return_exceptions=True,
    )
    respuestas = []
    for resultado in resultados:
        if isinstance(resultado, Exception):
            respuestas.append({"estado": "error", "detalle": str(resultado)})
        else:
            respuestas.append(resultado)
    return respuestas


print(asyncio.run(consultar_fuentes()))

Salida esperada:

[{'fuente': 'notas', 'estado': 'ok'}, {'estado': 'error', 'detalle': 'calendario no responde'}]

return_exceptions=True permite conservar los resultados válidos y tratar cada fallo por separado. Úsalo solo cuando el caso de uso permita una respuesta parcial; si todas las fuentes son obligatorias, deja que la excepción suba y manéjala en la frontera adecuada.

Errores frecuentes¶

• Olvidar await y trabajar con la coroutine sin ejecutar.
• Usar asyncio.run dentro de un bucle que ya está activo.
• Confundir concurrencia con ejecutar cálculos en paralelo.
• Usar time.sleep, una petición síncrona o un bucle pesado dentro de una coroutine y bloquear a todas las demás.
• Crear clientes HTTP repetidamente sin cerrarlos.
• Capturar Exception y convertir un fallo total en éxito sin informar.
• Usar return_exceptions=True sin revisar cada resultado.
• Medir velocidad con sleep(0) y sacar conclusiones sobre un sistema real.

Práctica de transferencia¶

1. Convierte una función que espere con time.sleep a una coroutine.
2. Ejecuta tres consultas simuladas con gather y conserva los resultados en el mismo orden que las fuentes.
3. Añade un timeout y decide qué respuesta es correcta si una fuente tarda.
4. Provoca una excepción en una de las consultas y compara gather con y sin return_exceptions=True.
5. Explica por qué una función que calcula millones de operaciones no se acelera automáticamente por ser async.

Producto evaluable¶

Crea 12_asyncio.py como script o cuaderno marimo. Debe:

1. Consultar al menos tres fuentes simuladas con async def y gather.
2. Mostrar resultados en un orden estable y distinguir éxito, error y timeout.
3. Cerrar los clientes o recursos asíncronos con async with si usas la ampliación del cliente HTTP.
4. Incluir una celda Markdown que explique por qué el caso es concurrente y no necesariamente paralelo.
5. Comparar una operación asíncrona con una función síncrona que se ejecuta con asyncio.to_thread.

Criterio de aceptación: la práctica se repite sin red ni claves, no deja coroutines sin esperar y trata explícitamente al menos un fallo.

Formato de entrega: script o cuaderno marimo ejecutado, con salida esperada, interpretación y conclusión.

Resumen y referencia rápida¶

Siguiente: 1.17 · Type hints y testing con pytest.