Sintaxis async y await en Python para concurrencia eficiente

Aprende a convertir un flujo bloqueante en concurrencia eficiente con async y await en Python. Con una refactorización guiada, verás cómo reemplazar llamadas que bloquean por corrutinas, usar asyncio.gather para paralelizar y ejecutar todo con asyncio.run. Resultado: de un total de 9 s en versión síncrona a cerca de 3 s en versión asíncrona, con tareas completando según su duración.

¿Qué son async y await en Python moderno?

Async define funciones asíncronas, corrutinas, que pueden pausar su ejecución sin bloquear el programa. Await pausa una corrutina hasta que otra operación asíncrona termina. Al combinarlas, se obtiene concurrencia clara y eficiente.

• Corrutinas: funciones declaradas con async que pueden suspenderse y reanudarse.
• No bloqueo: al usar await correctamente no se congela el programa completo.
• Event loop: ciclo que orquesta corrutinas; si introduces llamadas bloqueantes, lo detienes.
• Beneficio medible: tiempos totales mucho menores cuando las esperas se solapan.

¿Cómo transformar código síncrono a asíncrono con asyncio?

La base es un flujo que procesa una lista de tareas con nombre y segundos de espera, imprime logs de inicio y fin, y mide tiempos. En su forma síncrona, el total fue de 9 segundos. Tras el refactor, con asyncio, la aplicación completa tardó alrededor de 3 segundos, y la tarea con 1 s de espera terminó antes que las demás.

¿Cuál es la base síncrona con time.sleep?

import time

def tarea(nombre, segundos):
    print(f"inicia {nombre}")
    time.sleep(segundos)
    print(f"finaliza {nombre}")
    return nombre, segundos

def main():
    # A, B, C, D con distintas duraciones
    tareas = [("A", 3), ("B", 2), ("C", 1), ("D", 3)]
    resultados = []
    for n, s in tareas:
        resultados.append(tarea(n, s))
    print(resultados)

if __name__ == "__main__":
    main()

¿Cómo convertir funciones en corrutinas con async?

• Cambia las funciones a corrutinas con async.
• Reemplaza time.sleep por await asyncio.sleep(segundos) para no bloquear.

import asyncio

async def tarea(nombre, segundos):
    print(f"inicia {nombre}")
    await asyncio.sleep(segundos)
    print(f"finaliza {nombre}")
    return nombre, segundos

¿Cómo ejecutar tareas en paralelo con asyncio.gather?

• Convierte main en corrutina.
• Usa await asyncio.gather(...) para correr varias corrutinas en paralelo.
• Ejecuta el event loop con asyncio.run(main()).

import asyncio

async def main():
    tareas = [
        tarea("A", 3),
        tarea("B", 2),
        tarea("C", 1),
        tarea("D", 3),
    ]
    resultados = await asyncio.gather(*tareas)
    print(resultados)

if __name__ == "__main__":
    asyncio.run(main())

• Paralelismo cooperativo: cada await cede el control al event loop.
• Orden de finalización: completa primero quien espera menos tiempo.
• Datos clave: versión síncrona ~9 s; versión asíncrona ~3 s.

¿Qué cuidar del event loop y qué reto practicar?

Si no usas await donde corresponde, bloqueas el event loop y el comportamiento se parece al flujo síncrono. Evita llamadas bloqueantes como time.sleep dentro de corrutinas; usa await asyncio.sleep y ejecuta todo con asyncio.run.

• Usa await en operaciones asíncronas para no bloquear.
• Reemplaza time.sleep por await asyncio.sleep dentro de corrutinas.
• Orquesta varias corrutinas con asyncio.gather.
• Arranca el event loop con asyncio.run.

Reto propuesto: crea una función asíncrona que simule pedir datos a una API usando await asyncio.sleep como delay de red, y pruébala en el intérprete interactivo.

import asyncio

async def obtener_datos():
    print("consultando API...")
    await asyncio.sleep(1)
    print("listo")
    return {"status": "ok"}

async def main():
    datos = await obtener_datos()
    print(datos)

asyncio.run(main())

¿Qué caso de uso te gustaría optimizar con asyncio? comparte tu idea en los comentarios.