Tutorial ‘asyncio’ para el programador con el tiempo en contra.

Introducción

En el mundo en constante evolución de la programación en Python, entender la biblioteca asyncio se está volviendo cada vez más importante. asyncio permite la programación asíncrona, concurrente y paralela en un lenguaje que tradicionalmente era síncrono y de un solo hilo.

A medida que las aplicaciones de Python crecen en complejidad y dependen de tareas limitadas por E/S como solicitudes web o llamadas a bases de datos, es crucial gestionar estas tareas de manera eficiente. Ahí es donde brilla asyncio. Sin embargo, comenzar con asyncio puede ser desalentador, especialmente si no estás familiarizado con las corrutinas, tareas, futuros y bucles de eventos. Este tutorial tiene como objetivo guiarte a través de los conceptos básicos de asyncio, completo con ejemplos funcionales.

Al final de este tutorial, tendrás un sólido conocimiento de asyncio y estarás listo para implementarlo en tus proyectos, ya sea que estés construyendo un microservicio o una aplicación de interfaz gráfica de usuario.

Corrutinas

Una corrutina es una generalización de una subrutina que se puede pausar y reanudar, permitiendo que otro código se ejecute durante las pausas. Las corrutinas son una parte clave de asyncio y nos permiten escribir código asíncrono en un estilo procedural.

En Python, las corrutinas se definen con la sintaxis async def:

async def my_coroutine():
    print("Hello, Coroutine!")

Para ejecutar la corrutina, no puedes simplemente llamarla como una función normal. Debes usar la palabra clave await.

await my_coroutine()

Vamos a simular una tarea de larga duración con la función sleep de asyncio.

import asyncio

async def my_coroutine():
    await asyncio.sleep(1)  # sleep for 1 second
    print("Hello, Coroutine!")

async def main():
    await my_coroutine()

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

Administradores de contexto asíncronos e iteradores asíncronos

En aplicaciones del mundo real, a menudo necesitas gestionar recursos como flujos de archivos, conexiones de red o conexiones a bases de datos. El uso de gestores de contexto e iteradores de forma asíncrona puede ser muy útil para tales tareas. Afortunadamente, asyncio de Python ofrece soporte nativo para gestores de contexto asíncronos e iteradores asíncronos, lo que hace que tu código no solo sea más eficiente, sino también más limpio y fácil de mantener.

Usando async with

La declaración async with se utiliza para invocar un gestor de contexto asíncrono. Por ejemplo, al trabajar con operaciones de archivos asíncronas, puedes leer un archivo sin bloquear el bucle de eventos de esta manera:

import aiofiles
import chardet
import asyncio

async def read_large_file():
    async with aiofiles.open('large_file.txt', mode='rb') as f:
        raw_data = await f.read()

    detected = chardet.detect(raw_data)
    encoding_type = detected['encoding']
    contents = raw_data.decode(encoding_type)
    print(contents)

async def main():
    await read_large_file()

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

Un ejemplo más:

# Async context managers for managing resources
async with aiohttp.ClientSession() as session:
    async with session.get('https://api.example.com/data') as resp:
        data = await resp.json()

Best Practice: Siempre prefiera usar async with para adquirir y liberar recursos.

Usando async for

La declaración async for se puede utilizar con un objeto que es un iterador asíncrono. Por ejemplo, para obtener múltiples URL de forma asíncrona:

import asyncio

# Define an asynchronous generator simulating fetching data from URLs
async def fetch_data_from_urls():
    urls = ["https://www.google.com", "https://www.apple.com", "https://www.microsoft.com"]
    for url in urls:
        await asyncio.sleep(2)  # Simulate network delay
        yield f"Fetched data from {url}"

# Define a coroutine that uses the asynchronous generator
async def process_fetched_data():
    async for data in fetch_data_from_urls():
        print(f"Processing: {data}")

# Run the asynchronous event loop
asyncio.run(process_fetched_data())

Tanto los bucles for como async for en Python tienen propósitos similares: se utilizan para iterar sobre elementos. Sin embargo, no son intercambiables y están diseñados para diferentes tipos de iterables.

• El bucle for se utiliza para iterar sobre iterables síncronos, como listas, o generadores síncronos.

for item in [1, 2, 3]:
    print(item)

• El bucle async for está diseñado para iterar sobre iterables asíncronos o generadores asíncronos.

async for item in async_generator():
    print(item)

La diferencia clave es que no puedes usar un bucle for estándar para iterar sobre iterables asíncronos o generadores asíncronos. Del mismo modo, no puedes usar un bucle async for para iterar sobre iterables o generadores síncronos tradicionales.

Si intentas mezclarlos, Python generará un TypeError.

Best Practice: Utiliza async for al trabajar con iteradores asíncronos para hacer que tu código sea más legible y fácil de mantener.

Tareas y Concurrencia

Las tareas se utilizan para programar corrutinas de manera concurrente. Cuando una corrutina se envuelve en una Task con funciones como asyncio.create_task(), la corrutina se programa automáticamente para ejecutarse pronto.

import asyncio

async def my_coroutine():
    await asyncio.sleep(1)
    print("Hello, Coroutine!")

async def main():
    # Create a task out of a coroutine
    task = asyncio.create_task(my_coroutine())
    # wait for the task to finish
    await task

# Run the main coroutine
asyncio.run(main())

task = asyncio.create_task(my_coroutine())

RuntimeError: no running event loop

asyncio.run(my_coroutine())

Las tareas pueden cancelarse si ya no son necesarias. Esto se hace con el método cancel en el objeto Task. También puedes establecer tiempos de espera para las tareas utilizando la función asyncio.wait_for:

import asyncio

async def my_coroutine():
    await asyncio.sleep(10)
    print("Hello, Coroutine!")

async def main():
    # Create a task out of a coroutine
    task = asyncio.create_task(my_coroutine())

    # Cancel the task
    task.cancel()

    # Set a timeout for a task
    try:
        await asyncio.wait_for(my_coroutine(), timeout=1.0)
    except asyncio.TimeoutError:
        print("Task took too long!")

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

Tareas, corrutinas y futuros son todos tipos de objetos esperables (awaitables). Un objeto esperable es algo que puedes usar en una expresión await. Los futuros son objetos esperables de nivel más bajo que representan un resultado eventual de un cálculo.

Medir el tiempo de ejecución en corrutinas con decoradores

Puedes usar decoradores para medir el tiempo de ejecución de tus corrutinas. Aquí tienes un ejemplo:

import asyncio
import time

def timer(func):
    async def wrapper(*args, **kwargs):
        start_time = time.time()
        result = await func(*args, **kwargs)
        end_time = time.time()
        print(f"Time taken: {end_time - start_time} seconds")
        return result
    return wrapper

@timer
async def my_coroutine():
    await asyncio.sleep(1)
    print("Hello, Coroutine!")

async def main():
    await my_coroutine()

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

¿Por qué medir el tiempo de ejecución?

Antes de sumergirse en el código, es crucial entender por qué medir el tiempo de ejecución de tus corrutinas es beneficioso en primer lugar. Aquí hay algunas razones:

1. Optimización del rendimiento: Uno de los principales beneficios es identificar cuellos de botella o problemas de rendimiento en tu código. Si una corrutina en particular tarda demasiado en ejecutarse, podría convertirse en un candidato para optimización.
2. Asignación de recursos: Comprender la complejidad temporal de tus tareas puede ayudar en una mejor asignación de recursos. Por ejemplo, podrías decidir descargar una tarea que consume mucho tiempo a un hilo de trabajo o a otra máquina.
3. Depuración: Ocasionalmente, las tareas pueden colgarse o entrar en un bucle infinito. Medir el tiempo de ejecución puede servir como una herramienta de depuración para identificar tales problemas.
4. Experiencia del usuario: Para aplicaciones orientadas al usuario, el tiempo de respuesta es crítico. Saber cuánto tiempo tardan las tareas puede ayudarte a tomar decisiones basadas en datos para mejorar la experiencia del usuario. Por ejemplo, podrías introducir una pantalla de carga para tareas que tardan más de unos segundos.
5. Escalabilidad: Al construir un sistema que necesita escalar, es esencial entender cómo se desempeñan los componentes individuales. Medir el tiempo de ejecución de tus corrutinas te brinda información sobre cómo se desempeñará tu aplicación bajo diferentes cargas.

Al estar atento al tiempo que tarda cada corrutina en ejecutarse, no solo estás escribiendo código, sino también asegurándote de que sea eficiente, escalable y proporcione una experiencia de usuario fluida.

Problemas con Corrutinas y Tareas

Un problema potencial con las corrutinas y las tareas es que pueden ser difíciles de depurar. Los errores podrían no generarse hasta que se espera la corrutina, lo que puede dificultar la localización del origen del problema.

Manejo de errores diferido en corrutinas

Es importante tener en cuenta que los errores en una corrutina no se generarán hasta que se espere la corrutina. Esto se debe a que las corrutinas son esencialmente ‘perezosas’, lo que significa que no hacen ningún trabajo hasta que les pides explícitamente que lo hagan usando await. Esto puede ser una espada de doble filo: por un lado, proporciona más control sobre cuándo se ejecuta el código, pero por otro, puede dificultar la depuración ya que los errores se aplazan.

Solución y Workaround:

• Usando gather con return_exceptions: Si estás ejecutando múltiples corrutinas de forma concurrente y deseas capturar todas las excepciones, considera usar asyncio.gather con el parámetro return_exceptions=True. De esta manera, las excepciones se devolverán en lugar de generarse, lo que te permitirá manejarlas de manera elegante.

import asyncio

async def problematic_coroutine():
    raise ValueError("Some error")

async def another_problematic_coroutine():
    raise KeyError("Another error")

async def main():
    results = await asyncio.gather(
        problematic_coroutine(),
        another_problematic_coroutine(),
        return_exceptions=True
    )

    for result in results:
        if isinstance(result, Exception):
            print(f"Caught an error: {result}")

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

El bucle de eventos en asyncio

El bucle de eventos es el núcleo de cada aplicación asyncio. Puedes acceder y gestionar el bucle de eventos manualmente, aunque en muchos casos, las API de alto nivel de asyncio son todo lo que necesitas.

Así es como puedes acceder al bucle de eventos y usarlo para ejecutar una corrutina:

import asyncio

async def my_coroutine(n):
    print(f"Coroutine {n} starting")
    await asyncio.sleep(n)  # simulate IO-bound task with sleep
    print(f"Coroutine {n} completed")

# Get the current event loop
loop = asyncio.get_event_loop()

# Create multiple tasks to run
tasks = [loop.create_task(my_coroutine(i)) for i in range(1, 4)]

# Gather tasks and run them
loop.run_until_complete(asyncio.gather(*tasks))

loop.close()

En este script, definimos una función asíncrona my_coroutine que simula una tarea limitada por E/S que consume tiempo durmiendo durante un número de segundos dado por n. Luego obtenemos el bucle de eventos actual y lo usamos para crear una serie de tareas. Cada tarea es una instancia de my_coroutine con un argumento diferente para n.

Finalmente, usamos asyncio.gather para combinar estas tareas en un solo objeto esperable, y usamos run_until_complete para ejecutar este objeto. El script imprime un mensaje cuando cada corrutina comienza y se completa, para que puedas ver que las corrutinas se ejecutan de manera concurrente.

Recuerda, asyncio está diseñado para tareas limitadas por E/S y no para tareas limitadas por CPU. Si intentas usar asyncio con tareas limitadas por CPU, no obtendrás un verdadero paralelismo, porque asyncio de Python se basa en corrutinas y un bucle de eventos, que es una arquitectura de un solo hilo. Si tienes tareas limitadas por CPU, considera usar hilos o multiprocesamiento en Python.

Además, cuando terminamos con un bucle, es una buena práctica cerrarlo (close()).

Modo ‘Debug’

El modo de depuración de asyncio puede proporcionar información más detallada sobre tu código asyncio y ayudarte a diagnosticar problemas. Para habilitar el modo de depuración, puedes establecer la variable de entorno PYTHONASYNCIODEBUG en 1, o puedes habilitarlo en tu código.

asyncio realiza una comprobación para PYTHONASYNCIODEBUG durante la importación del módulo. Por lo tanto, necesitas configurar la variable de entorno antes de la primera importación de asyncio.

import os
os.environ['PYTHONASYNCIODEBUG'] = '1'
import logging

logging.basicConfig(level=logging.DEBUG)

When debug mode is enabled, asyncio logs more events and checks for common mistakes, such as blocking the event loop.

import os
os.environ['PYTHONASYNCIODEBUG'] = '1'
import asyncio
import time

async def my_coroutine(n):
    print(f"Coroutine {n} starting")
    await asyncio.sleep(n)
    print(f"Coroutine {n} completed")

async def blocking_coroutine():
    print("Blocking coroutine starting")
    time.sleep(2)  # Blocking call
    print("Blocking coroutine completed")

async def main():
    tasks = [
        asyncio.create_task(my_coroutine(1)),
        asyncio.create_task(my_coroutine(2)),
        asyncio.create_task(blocking_coroutine())
    ]
    await asyncio.gather(*tasks)

# Get the current event loop
loop = asyncio.get_event_loop()

# Enable debug mode
loop.set_debug(True)

# Run the main coroutine
try:
    loop.run_until_complete(main())
finally:
    loop.close()

En este ejemplo, definimos dos funciones asíncronas: my_coroutine y blocking_coroutine. La función my_coroutine es similar a la del ejemplo anterior, mientras que blocking_coroutine está diseñada para hacer una llamada bloqueante con time.sleep().

También creamos una corrutina principal que crea tareas tanto para my_coroutine como para blocking_coroutine. Luego ejecutamos la corrutina principal usando loop.run_until_complete().

En este caso, la función blocking_coroutine bloqueará el bucle de eventos, lo cual es un error común al usar asyncio. Al ejecutar este script en modo de depuración, asyncio imprimirá un mensaje de advertencia para notificarte que el bucle de eventos está siendo bloqueado por la función blocking_coroutine.

Recuerda habilitar el modo de depuración en producción con precaución, ya que puede ralentizar tu aplicación debido a las comprobaciones adicionales. Es mejor usarlo durante el desarrollo y las pruebas para detectar cualquier problema potencial con tu código asyncio.

Conclusión

¡Bien hecho al concluir este tutorial sobre los conceptos básicos de asyncio! Ahora tienes un sólido conocimiento de los principios fundamentales de asyncio, desde corrutinas y tareas hasta depuración y gestión del bucle de eventos.

Pero hay más en asyncio. Su verdadero potencial se despliega al gestionar flujos de trabajo complejos y coordinar entre diferentes tareas. Aquí es donde entran en juego temas avanzados como primitivas de sincronización, colas, semáforos y transportes y protocolos.

En nuestro próximo artículo, profundizaremos en estos conceptos avanzados de asyncio, equipándote con un conjunto de habilidades integral para enfrentar desafíos complejos de programación asíncrona en Python. ¡Continúa practicando, explorando y nos vemos en nuestro próximo tutorial para más aventuras de aprendizaje!

Etiquetas: asynchronous programming