Python: asyncio – Cheatsheet

En el mundo actual, donde la velocidad y la eficiencia son claves, la programación asíncrona se ha convertido en una herramienta fundamental para aprovechar al máximo los recursos computacionales. Python, uno de los lenguajes de programación más populares, ofrece una poderosa librería para el desarrollo asíncrono: asyncio.

He creado este post porque estoy aprendiendoa usarlo y quiero les sirva a ustedes y a mi como Cheatsheet. Es probable que lo actualice de vez en cuando con nuevos ejemplos de código.

¿Qué es asyncio?

asyncio es una librería estándar de Python que facilita la escritura de código asíncrono. Permite ejecutar múltiples tareas de forma concurrente, aprovechando al máximo el tiempo de espera y la capacidad de procesamiento del sistema.

¿Por qué usar asyncio?

Las ventajas de utilizar asyncio son numerosas:

• Mejora el rendimiento: Al ejecutar tareas de forma concurrente, se reduce el tiempo total de ejecución, especialmente en operaciones que esperan respuestas de Entrada/Salida, ya sea a Base de Datos, APIs o archivos.
• Escala eficientemente: La naturaleza asíncrona de asyncio permite manejar un gran número de conexiones simultáneas sin consumir demasiados recursos.
• Mayor capacidad de respuesta: La interfaz de usuario se mantiene receptiva mientras se ejecutan tareas en segundo plano.
• Código más limpio: La sintaxis de asyncio facilita la escritura de código modular y legible.

¿Cómo funciona asyncio?

El núcleo de asyncio es el Event Loop. Este loop monitorea continuamente los eventos y ejecuta las tareas que están listas para ser ejecutadas. Las tareas se definen como corrutinas, funciones especiales que pueden suspenderse y reanudarse posteriormente.

Términos Clave

• Coroutines(corrutinas:) Es el Building Blocks de asyncio. Defina funciones asíncronas usando la sintaxis async def. Las corrutinas no se ejecutan de inmediato, necesitan ser programadas.
• Event Loops: El núcleo de asyncio, responsable de gestionar la ejecución de las corrutinas. Se usa asyncio.get_event_loop() para obtener el loop.
• Tasks: Wrapper para corrutinas, creadas con asyncio.create_task(). Las Tasks gestionan la programación y el estado de las corrutinas que se ejecutan en el Event Loops.
• await: Palabra clave utilizada dentro de las corrutinas para pausar la ejecución hasta que otra corrutina se complete. El valor devuelto por la expresión await es el resultado de la corrutina esperada.

Funciones Clave

• asyncio.run() La función de nivel superior para iniciar el bucle de eventos y ejecutar una corrutina. Normalmente el punto de entrada de los programas asyncio.
• asyncio.create_task() Programa una corrutina para que se ejecute en el bucle de eventos, devolviendo un objeto Task.
• asyncio.sleep() Una corrutina para introducir retardos en la ejecución, cediendo el control al bucle de eventos.
• asyncio.gather() Ejecuta múltiples corrutinas de forma concurrente, devolviendo sus resultados en una lista. Es ideal para ejecutar tareas asíncronas que no dependen unas de otras.
• asyncio.wait_for() Espera a que una corrutina se complete, con un tiempo de espera opcional.
• asyncio.as_completed() Produce corrutinas en el orden en que se completan.

Primitivas de Sincronización

• asyncio.Semaphore Limita el número de accesos concurrentes a un recurso compartido.
• asyncio.Lock Un candado de exclusión mutua básico.
• asyncio.Event Mecanismo de comunicación entre corrutinas.

Códigos de Ejemplo

import asyncio

async def fetch_data(url):
# Simula una solicitud de red
await asyncio.sleep(1)
return f"Datos de {url}"

async def main():
result1 = await fetch_data("https://api.example.com/1")
result2 = await fetch_data("https://api.example.com/2")
print(result1, result2)

asyncio.run(main())

Ejemplo de Semaphore

Los semáforos permiten controlar el número de procesos que pueden acceder a un recurso simultáneamente. Se inicializan con un valor que indica el número de permisos disponibles.

import asyncio

# Semáforo con 2 permisos
semaphore = asyncio.Semaphore(2)

async def acceso_recurso(nombre):
    # Adquiere un permiso
    await semaphore.acquire()

    # Simula acceso al recurso
    await asyncio.sleep(1)
    print(f"{nombre} accediendo al recurso")

    # Libera un permiso
    semaphore.release()

async def main():
    # Crea tareas que intentan acceder al recurso
    tarea1 = asyncio.create_task(acceso_recurso("Tarea 1"))
    tarea2 = asyncio.create_task(acceso_recurso("Tarea 2"))
    tarea3 = asyncio.create_task(acceso_recurso("Tarea 3"))

    # Espera a que todas las tareas terminen
    await asyncio.gather(tarea1, tarea2, tarea3)

asyncio.run(main())

Ejemplo de Lock

import asyncio

# Crea un candado
lock = asyncio.Lock()

async def acceso_recurso(nombre):
    # Adquiere el candado
    await lock.acquire()

    # Simula acceso al recurso
    await asyncio.sleep(1)
    print(f"{nombre} accediendo al recurso")

    # Libera el candado
    lock.release()

async def main():
    # Crea tareas que intentan acceder al recurso
    tarea1 = asyncio.create_task(acceso_recurso("Tarea 1"))
    tarea2 = asyncio.create_task(acceso_recurso("Tarea 2"))
    tarea3 = asyncio.create_task(acceso_recurso("Tarea 3"))

    # Espera a que todas las tareas terminen
    await asyncio.gather(tarea1, tarea2, tarea3)

asyncio.run(main())

Ejemplo de TaskGroup

Un administrador de contexto asincrónico que contiene un grupo de tareas. Las tareas se pueden agregar al grupo usando create_task(). Todas las tareas se esperan (awaited) cuando sale el administrador de contexto.

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(some_coro(...))
        task2 = tg.create_task(another_coro(...))
        
    print(f"Both tasks have completed now: {task1.result()}, {task2.result()}")

Ejemplo de Timeout

Return an asynchronous context manager that can be used to limit the amount of time spent waiting on something.

delay can either be None, or a float/int number of seconds to wait. If delay is None, no time limit will be applied; this can be useful if the delay is unknown when the context manager is created.

async def main():
    try:
        async with asyncio.timeout(10):
            await long_running_task()
    except TimeoutError:
        print("The long operation timed out, but we've handled it.")

    print("This statement will run regardless.")

Si long_running_task tarda más de 10 segundos en completarse, el administrador de contexto cancelará la tarea actual y manejará el asyncio.CancelledError resultante internamente, transformándolo en un TimeoutError que se puede detectar y manejar.

Ejemplo de wait_for()

La función asyncio.wait_for() permite esperar a que una corrutina termine, con un tiempo de espera opcional.

import asyncio

async def tarea_larga():
    # Simula una tarea que tarda 3 segundos en completarse
    await asyncio.sleep(3)
    return "Resultado de la tarea larga"

async def main():
    try:
        # Espera a que la tarea termine en 2 segundos
        resultado = await asyncio.wait_for(tarea_larga(), timeout=2)
        print(f"Resultado: {resultado}")
    except asyncio.TimeoutError:
        print("La tarea tardó demasiado tiempo")

asyncio.run(main())

Ejemplo de as_completed()

La función asyncio.as_completed() crea un iterable que produce corrutinas a medida que se completan.

import asyncio

async def tarea_1():
    # Simula una tarea que tarda 1 segundo en completarse
    await asyncio.sleep(1)
    return "Resultado de la tarea 1"

async def tarea_2():
    # Simula una tarea que tarda 2 segundos en completarse
    await asyncio.sleep(2)
    return "Resultado de la tarea 2"

async def main():
    # Crea un iterable de corrutinas
    tareas = [tarea_1(), tarea_2()]

    # Recorre las corrutinas a medida que se completan
    for futuro in asyncio.as_completed(tareas):
        resultado = await futuro
        print(f"Resultado: {resultado}")

asyncio.run(main())