AWS asíncrono en python

Recientemente me encontré con dos librerías muy interesantes que permiten utilizar algunos servicios de AWS utilizando el asincronismo, para utilizar sus servicios AWS expone una API por la cual se podría utilizar los servicios utilizando librerías como httpx para hacer peticiones a servicios de forma asíncrona, sin embargo básicamente al hacer esto estarías creando casi que un "mini framework" o un mini "boto3" pero asíncrono y generalmente no hay mucho tiempo para realizar este trabajo, por ello me di a la tarea de buscar algunas herramientas y encontré dos las cuales me parecieron muy prometedoras y de las cuales hablaré a continuación

Por que Asíncrono?

Trabajar con el concepto de tareas asíncronas se resume en utilizar de forma eficiente los recursos en sus tiempos de ocio, es decir evitar que los recursos computacionales se queden sin nada que hacer, esto permite atender grandes cantidades de tareas de la forma más eficiente posible, la mayor ventaja que se puede obtener aquí es en los procesos de I/O es decir en las operaciones que involucren escritura de disco duro, acceso red y consultas a bases de datos, para nuestro caso para comunicarnos con AWS se utiliza servicios de red y lectura de archivos por lo tanto es un caso que aplica a la perfección, en python las librerías nacen nativamente síncronas por lo tanto la librería boto3 que es la librería más usada con AWS realiza todas las peticiones sin utilizar una gestión asíncrona ni utilizar el eventloop, si vienes de javascript seguro este concepto te es muy familiar este se explica en la Figura 1

Figura 1: Asincronismo en javascript

Asyncio

Este es una librería/framework que se agregó en python a partir de la versión 3.4 para trabajar con coroutines (en la versión 3.5 se agrega async y await) internamente esta librería contiene el eventloop que se usará para "encolar" las tareas en el loop de eventos y notificar eventualmente, es de aclarar que el objetivo de utilizar asyncio no es que el código "funcione más rápido"(quizás si no tener tantos problemas como los threads ver Figura 2). Desde un principio asyncio tenía (tiene) muchas cosas en su documentación lo que dificultad un poco entender como funciona de manera sencilla, en esta charla de Yury Selivanov se deja un poco más claro el tema, por lo tanto siguiendo la idea prinicpal de la charla hay algunas funciones que a nosotros como usuarios finales nos deben de interesar, y para seguir este artículo usaremos sólo dos

Iniciar el event loop de asyncio
Llamar el async/await functions

Figura 2 Threading mas problemas que soluciones (it is only a joke) Obtenida de https://www.reddit.com/r/ProgrammerHumor/comments/dtiufv/multithreading_fixing_a_problem/

Librerías

He encontrado dos muy buenas librerías que tienen muy buena pinta, sin embargo están creciendo y aun no tienen todos los servicios de AWS integrados, las librerías son aioaws y aioboto3 si has trabado con boto3 y quieres mantener una sintaxis y métodos similares esa es tu opción, aioaws es hecho desde 0 y es un proyecto del mismo creador de pydantic, esta librería contiene type hints lo cual la hace bastante interesante, sin embargo recalco de nuevo que ninguna de las dos librerías aun integran toda la suite de AWS, en dado caso que necesites utilizar varios servicios lo mas recomendable seria utilizar llamados a la API de AWS utilizando httpx, en este artículo veremos la herramienta aioaws. Para instalar ejecutamos el siguiente comando

pip install aioaws

Ejemplos aioaws

Esta es una librería realizada desde 0 sin ninguna dependencia a boto3, utilizando httpx, aiofile y pydantic crea una base sólida para construir un SDK robusto para AWS utilizando asincronismo con asyncio. Veamos los siguientes ejemplos trabajando con AWS S3

Subir archivos

import asyncio
import aiofiles
import os
from aioaws.s3 import S3Client, S3Config
from httpx import AsyncClient
from dotenv import load_dotenv

load_dotenv(".env")


async def upload(client: AsyncClient, file: bytes):
    s3 = S3Client(
        client,
        S3Config(
            aws_access_key=os.getenv("AWS_ACCESS_KEY"),
            aws_secret_key=os.getenv("AWS_SECRETS_KEY"),
            aws_s3_bucket="bucket-post",
            aws_region=os.getenv("AWS_REGION"),
        ),
    )

    await s3.upload("myfiles/dev.jpg", file)


async def open_file(location: str) -> bytes:
    async with aiofiles.open(location, "rb") as file:
        content = await file.read()
    return content


async def main():
    async with AsyncClient() as client:
        file_to_upload = await open_file("dev.jpg")
        await upload(client, file_to_upload)
        print("Uploaded Successfully")


asyncio.run(main())

Ejecuto el codigo y obtengo el siguiente resultado

Uploaded Successfully

Figura 3: Archivo subido a el bucket de AWS

En el código anterior la funcion upload es el encargado enviar la información a S3, se utilizan dos clases clave S3CLient y S3Config en donde estas tendrán las credenciales de acceso, utilizando dotenv se cargan estas credenciales desde un archivo .env y con el objeto s3 ya se llama al método upload(), es importante aclarar que el archivo se debe abrir utilizando aiofiles para que no quede en blocking el proceso, por esto la función open_file() utiliza el aiofiles para abrir un archivo y lo lee agregando await

Listar archivos

import asyncio
import aiofiles
import os
from aioaws.s3 import S3Client, S3Config
from httpx import AsyncClient
from dotenv import load_dotenv, find_dotenv

load_dotenv(find_dotenv())


async def get_files(client: AsyncClient):
    s3 = S3Client(
        client,
        S3Config(
            aws_access_key=os.getenv("AWS_ACCESS_KEY"),
            aws_secret_key=os.getenv("AWS_SECRETS_KEY"),
            aws_s3_bucket="bucket-post",
            aws_region=os.getenv("AWS_REGION"),
        ),
    )
    files = [file async for file in s3.list()]
    print(files)





async def main():
    async with AsyncClient() as client:
        await get_files(client)


asyncio.run(main())

ejecuto el codigo y obtengo el siguiente resultado

[S3File(key='myfiles/dev.jpg', last_modified=datetime.datetime(2021, 10, 16, 18, 0, 59, tzinfo=datetime.timezone.utc), size=176347, e_tag='0d0f00f387de7e9dccc960aa323fd67a', storage_class='STANDARD')]

una lista de objetos de tipo S3File que es una clase que implementa la librería el cual tiene atributos como el key, fecha modificación, y la clase de tipo de almacenamiento, en el código anterior utilizamos la funcion get_files() para obtener todos los folder dentro del bucket, sin embargo al ser muchos archivos, traerlos todos de una vez puede generar una carga muy pesada por eso se utilizan iterators en este caso deben ser asíncronos por eso la siguiente línea de código

files = [file async for file in s3.list()]

esto es un list comprehension para iterar utilizando asincronismo

Url firmada

Esta es una de las opciones más utilizadas con los buckets pues la mayoría son privados y para ver algún documento se generan urls firmadas con acceso temporal, esta función está disponible como se aprecia a continuación

import asyncio
import aiofiles
import os
from aioaws.s3 import S3Client, S3Config
from httpx import AsyncClient
from dotenv import load_dotenv, find_dotenv

load_dotenv(find_dotenv())

async def generate_signed_url(client: AsyncClient):
    s3 = S3Client(
        client,
        S3Config(
            aws_access_key=os.getenv("AWS_ACCESS_KEY"),
            aws_secret_key=os.getenv("AWS_SECRETS_KEY"),
            aws_s3_bucket="bucket-post",
            aws_region=os.getenv("AWS_REGION"),
        ),
    )
    signed_url = s3.signed_download_url(path="myfiles/dev.jpg", max_age=120)
    print(signed_url)


async def main():
    async with AsyncClient() as client:
        await generate_signed_url(client)


asyncio.run(main())

Ejecuto y obtengo el siguiente resultado

https://bucket-post.s3.us-east-1.amazonaws.com/myfiles/dev.jpg?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=AKIAUHHFBM4CL7PDM5PE%2F20211016%2Fus-east-1%2Fs3%2Faws4_request&X-Amz-Date=20211016T182441Z&X-Amz-Expires=120&X-Amz-SignedHeaders=host&X-Amz-Signature=99ef9bc9b95bcc32fd86463e7432f4cad646ccada01c5db3108b5c526a5eeab2

En el codigo anterior dentro de la función generate_signed_url() se llama al método signed_download_url() recibe dos parámetros el key del archivo y el tiempo de vida de la url, en este caso son 120 segundos

Borrar archivos

import asyncio
import aiofiles
import os
from aioaws.s3 import S3Client, S3Config
from httpx import AsyncClient
from dotenv import load_dotenv, find_dotenv

load_dotenv(find_dotenv())


async def delete_file(client: AsyncClient, key: str):
    s3 = S3Client(
        client,
        S3Config(
            aws_access_key=os.getenv("AWS_ACCESS_KEY"),
            aws_secret_key=os.getenv("AWS_SECRETS_KEY"),
            aws_s3_bucket="bucket-post",
            aws_region=os.getenv("AWS_REGION"),
        ),
    )
    await s3.delete(key)
    print("File deleted successfully")


async def main():
    async with AsyncClient() as client:
        await delete_file(client, "myfiles/dev.jpg")


asyncio.run(main())

Ejecuto y obtengo el siguiente resultado

File deleted successfully

Figura 4: Archivo eliminado del bucket

En el código anterior es más sencillo se llama a la función delete_file() pasandole el key a eliminar y del objeto s3 se llama a delete() pasando como parámetro el key, este borra el archivo y no retorna información, adicional hay otros métodos que permiten borrar los archivos de todo una carpeta estos los pueden encontrar la documentación

Bonus: Descargar archivo

import asyncio
import aiofiles
import os
from aioaws.s3 import S3Client, S3Config
from httpx import AsyncClient
from dotenv import load_dotenv, find_dotenv
import httpx

load_dotenv(find_dotenv())


async def generate_signed_url(client: AsyncClient):
    s3 = S3Client(
        client,
        S3Config(
            aws_access_key=os.getenv("AWS_ACCESS_KEY"),
            aws_secret_key=os.getenv("AWS_SECRETS_KEY"),
            aws_s3_bucket="bucket-post",
            aws_region=os.getenv("AWS_REGION"),
        ),
    )
    signed_url = s3.signed_download_url(path="myfiles/dev.jpg", max_age=120)
    return signed_url




async def download_file(client: AsyncClient, key: str):
    signed_url: str = await generate_signed_url(client)
    name_file = key.split("/")[-1]
    async with client.stream("GET", signed_url) as response:
        async with aiofiles.open(f"aio{name_file}", mode="wb", mode="wb") as file:
            async for chunk in response.aiter_bytes():
                await file.write(chunk)


async def main():
    async with AsyncClient() as client:
        await download_file(client, "myfiles/dev.jpg")


asyncio.run(main())

Ejecuto el codigo y obtengo el siguiente resultado, el archivo lo renombre como aiodev.jpg ya que dev.jpg ya existía

Figura 5: Archivo descargado desde el bucket

En el código anterior utilizando una función .stream() de httpx que permite descargar el contenido de una url por método GET en bytes se itera utilizando el iterador de forma asíncrona, recordando que los iteradores en este contexto funcionan utilizando async antes del for, seguidamente utilizando aiofiles se escribe estos bytes en disco, este codigo de bonus es por si extrañas la función download_fileob() de boto3

Conclusión

Actualmente hay muchas librerías que ya están naciendo con asincronismo en python, en este campo existen muchas desventajas ya que al existir tantas librerías sincronas el trabajo para implementar frameworks y otras librerías asíncronas es un poco más difícil, sin embargo ya hay frameworks com FastAPI que nativamente funcionan de forma asíncrona y el concepto del eventloop lo esconden similar a como sucede en javascript creando una capa de abstracción para la complejidad que podría llegar a tener asyncio puro, finalmente mi idea era mostrar un ejemplo también de SES pero lo dejaré para otro artículo ya que este se puede tornar demasiado largo.