DEV Community: Eduardo Santos

Construyendo APIs Serverless Resilientes: Webhook Gateway con Amazon EventBridge

Eduardo Santos — Sat, 22 Feb 2025 17:51:28 +0000

Las arquitecturas distribuidas tienen un componente fundamental, el webhook se convirtió en ese componente para la integración de sistemas en tiempo real. Sin embargo, representa varios desafíos: ¿Cómo manejamos los picos de tráfico?, ¿Cómo garantizamos la entrega confiable de los mensajes?, ¿Qué sucede cuando los sistemas downstream están caídos?

Exploraremos como construir un webhook serverless utilizando los servicios de AWS, que no solo resuelve los desafíos anteriormente listados, sino que también nos proporciona escalabilidad y tolerancia a los fallos.

Es importante mencionar que con la propuesta de esta arquitectura permitiremos que no se sobrepase el soft limit de la ejecución de lambdas que tiene por defecto AWS (1000 funciones lambdas ejecutándose al mismo tiempo).

¿Por qué una arquitectura Serverless para webhooks?
Los webhooks por su naturaleza son de trafico impredecible, por lo cual, las arquitecturas Serverless son particularmente idóneas para este caso de uso. Asimismo, no es necesario mantener servidores a la espera de eventos para manejar los picos ocasionales de tráfico, y solo pagamos por lo que usamos.

Arquitectura por desarrollar:

API Gateway endpoint: Validara el payload y preparara los headers necesarios para llamar al EventBridge Bus.
EventBridge Bus: Distribuye los eventos enviados por el API Gateway hacia las reglas de eventbridge.
EventBridge Rule: Se evalúa cada evento recibido desde el EB Bus, y se determinan hacia donde debe de enviarse basado en los patrones del evento.
Cola SQS Standard: Actúa como un buffer y garantiza el procesamiento de mensajes y el manejo de picos de tráficos.
DLQ SQS Standard: Captura los eventos que fallan después de 3 reintentos, permite el análisis posterior de eventos fallidos y con posibilidad de reprocesamiento manual si es necesario.
Función Lambda: Procesa los mensajes de la cola SQS de forma asíncrona y escalable de los eventos.

Explicación del código
API Gateway endpoint

Source: Este campo se valida como prefijo en las reglas de eventbridge para saber cuales deben dejar pasar el evento.

EventBridge Rule

Suscripcion Función Lambda

Beneficios de esta Arquitectura

Resiliencia: El uso de SQS y DLQ garantiza que ningún mensaje se pierda
Escalabilidad: La arquitectura serverless escala automáticamente
Control de Costos: Solo pagamos por el procesamiento real
Mantenibilidad: Cada componente tiene una responsabilidad única
Observabilidad: Fácil monitoreo mediante CloudWatch

El código fuente completo lo podrás encontrar en el siguiente enlace: Repositorio de Github
Para el despliegue completo de esta solución se requiere tener:

Un bucket donde almacenar los stacks anidados.
Un parameter store con el nombre del bucket.
Los secretos necesarios de AWS para poder desplegar la solución.

Consumiendo Certificados de Secrets Manager en Ambientes Serverless

Eduardo Santos — Sat, 21 Sep 2024 20:49:55 +0000

Alguna vez has tenido la necesidad de utilizar certificados dentro de tu Lambda para poder autenticar peticiones que realizas a servidores o consumo de servicios externos?

Como desarrolladores cada día nos enfrentamos a diferentes retos que nos motivan a mejorar nuestra arquitectura o procesos internos para aplicar mejores practicas. Trabajar con certificados es uno de esos retos que podemos enfrentar en el día a día , y estos a su vez con los diferentes retos de seguridad, algunos prefieren incluir los certificados como archivos en el despliegue de la Lambda que es una opción valida pero que en el caso que el certificado venza o se desactive por lo solicitud del cliente o el proveedor del certificado, esto requerirá desplegar de nuevo todas las lambdas o funciones que utilicen dicho certificado.

Para hacer frente al escenario anterior, podemos utilizar un servicio de AWS Secrets Manager nos permite almacenar nuestros secretos (incluyendo certificados) en la nube de AWS y ser consumidos por los servicios como ser Lambdas, Glue, etc.

Ventajas:
Rotación de certificados de forma centralizada.
Replican de certificados en diferentes regiones.
Desacoplamiento de nuestros secretos con nuestras funciones.
Posibilidad de crear nuevos servicios que consuman los certificados.

Escenario:
Debes de conectarte a una API de tercero que requiere que se autentique el usuario con certificados como PFX, CRT y KEY. Los certificados son asignados a un empleado que puede ser rotado dentro de la empresa, por lo cual, los certificados pueden cambiar sin haber expirado. Antes que se expiren los certificados el ente emisor emitirá los nuevos certificados para que el equipo de ingeniería realice los cambios necesarios en sus sistemas.

Solución 1
Adjuntar los certificados en cada una de las diferentes Lambdas.
Pros:
Los certificados no pueden ser accedidos, desde fuera de las Lambdas que los contienen.

Contras:
Al cambiar el certificado sera necesario volver a desplegar todas las funciones donde se requiera el uso de los certificados. Asimismo, cada nuevo servicio requerirá que se adjunte los certificados cuando sean necesarios.

Solución 2
Almacenar los certificados en Secrets Manager y consumirlos en las Lambdas que lo requieran.
Pros:
Cifrado de los certificados en reposo
Rotación de certificados sin necesidad de afectar las funciones que lo necesitan.
Los certificados podrán ser almacenados de forma temporal en la memoria efímera de nuestra lambda, reduciendo la necesidad de extraerlos constantemente de Secrets Manager.

Contras:
Si no se tiene una política de usuarios bien definida donde se limiten los accesos completos a los diferentes servicios, cualquier usuario con acceso a la consola de AWS podría obtener los certificados.

Por las ventajas que ofrece la segunda opción obtendremos por desarrollarla, nuestra arquitectura quedaría de la siguiente forma:

Podrás encontrar el repositorio del código a la solución 2 en el siguiente enlace https://github.com/edsantoshn/aws-secrets-manager

¿Qué es más rápido y económico para convertir archivos en AWS: Polar o Pandas?

Eduardo Santos — Sat, 10 Aug 2024 21:51:13 +0000

Ambas ofrecen una amplia gama de herramientas y ventajas que nos pueden poner en duda cual de los dos escoger en algún momento. No se trata de cambiar todos los procesos de la empresa para que inicien a utilizar Polars o una “muerte” a Pandas (esto no va a pasar en el futuro inmediato). Se trata de conocer otras herramientas que nos pueden ayudar a reducir costos y tiempo en los procesos obteniendo iguales o mejores resultados.

Cuando utilizamos servicios en la nube priorizamos ciertos factores, dentro de los que se encuentran el costo de los mismos. Los servicios que utilizo para este proceso son AWS Lambda con el runtime de Python 3.10 y S3 para almacenar el archivo crudo y el archivo convertido en parquet.

La intención es obtener un archivo CSV como data cruda y procesarlo con pandas y polars con la intención de verificar cual de estas dos bibliotecas nos ofrece una mejor optimización de recursos como memoria y el peso del archivo resultante.

Pandas
Es una biblioteca de Python especializada en la manipulación y análisis de datos, esta escrito en C y su lanzamiento inicial fue en el 2008.

*Polars *
Es una biblioteca de Python y Rust especializada en la manipulación y análisis de datos que permite procesos paralelos y está escrito mayormente en Rust y su lanzamiento fue en el 2022.

La arquitectura del proceso:

El proyecto es algo sencillo como se muestra en la arquitectura: El usuario deposita un archivo CSV en work/pandas o work/porlas y automáticamente inicia el s3 trigger a procesar el archivo para convertirlo en parquet y depositarlo en processed.

En este pequeño proyecto utilice dos lambdas con la siguiente configuración:
Memoria: 2 GB
Memoria efímera: 2 GB
Tiempo de vida: 600 segundos

Requerimientos
Lambda con pandas: Pandas, Numpy y Pyarrow
Lambda con polars: Polars

El dataset utilizado para la comparación esta disponible en kaggle con el nombre de “Rotten Tomatoes Movie Reviews – 1.44M rows” o puede ser descargado desde aquí.

El repositorio completo está disponible en GitHub puede ser clonado aquí.

Tamaño o Peso
La lambda que utilizada Pandas requiere dos complementos mas para poder crear un archivo parquet en este caso es PyArrow y una versión en especifico de numpy para la versión de pandas que estaba utilizando. Como resultado obtuvimos una lambda con un peso o tamaño de 74.4 MB algo muy cercano al límite que nos permite AWS de peso de la lambda.

La lambda con Polars no requiere otro complemento como PyArrow lo que simplifica la vida y reduce el tamaño de la lambda a menos de la mitad. Como resultado nuestra lambda tiene un peso o tamaño de 30.6 MB comparado con la primera nos da una amplitud para instalar otras dependencias que podremos necesitar para nuestro proceso de transformación.

Performance

La lambda con Pandas fue optimizada para utilizar compresión luego de la primera versión, sin embargo, se analizó también su comportamiento.
Pandas
Se tardo 18 segundos en procesar el dataset y utilizo 1894 MB de memoria para procesar el archivo CSV y generar un archivo Parquet en comparación con las otras versiones fue la que mas tiempo y recursos utilizo.

Pandas + Compression
Agregando una línea de código permitió mejorar un poco comparada con la versión anterior (Pandas), se tardo 17 segundos en procesar el dataset y utilizo 1837 MB, lo cual, no representa una mejora significativa en procesamiento y tiempo computacional, pero si en el tamaño del archivo resultante.

Polars
Se tardo 12 segundos en procesar el mismo dataset y utilizo solamente 1462 MB, en comparación con las dos anteriores representa un ahorro de tiempo del 44.44% y un consumo de memoria inferior.

Tamaño del archivo de salida

Pandas
La lambda en la cual no se estableció un proceso de compresión nos genero un archivo parquet de 177.4 MB.

Pandas + Compression
Al configurar la compresión en la lambda no genero un archivo parquet de 121.1 MB. Una pequeña línea u opción nos ayudo a reducir el tamaño del archivo en un 31.74%. Considerando que no es un cambio significativo de código es una muy buena opción.

Polars
Polars nos genero un archivo de 105.8 MB que comprado con la primera versión de pandas representa un ahorro del 40.36% y un 12.63% contra la versión pandas con compression.

Conclusión
No es necesario cambiar todos los procesos internos que utilizan Pandas para que ahora utilicen Polars, sin embargo, es importante considerar que si hablamos de miles o millones de ejecuciones lambdas utilizar Polars nos ayudara no solo con el tiempo de despliegue si no que nos ayudara a tener un costo menor debido al cobro por tiempo que realiza AWS en los servicios Serverless como lo es Lambda.
Asimismo, cuando traducimos ese 40.36% en millones de archivo estamos hablando de GBs o TBs, algo que si tendría un impacto significativo dentro de Datalake o Dataware house o incluso en un almacen de archivos en frio.

La reducción con Polars no solo se limitaría a estos dos factores, porque afectaría mucho la salida de datos y/o objetos de AWS porque es un servicio que si tiene un costo.

Consulta Select en SQLAlchemy

Eduardo Santos — Fri, 21 Oct 2022 02:37:59 +0000

Seguimos con la serie de conectarnos a una base de datos mediante SQLAlchemy, en este caso utilizare variables de entorno almacenadas en un archivo .env para facilitar la distribución de nuestro programa y mejorar un poco la seguridad de nuestro servidor de base de datos.

Una excelente practica es modularizar nuestro código dejando un modulo para la conexión de la base de datos, otro modulo con los modelos o clases de la base de datos y las queries.

Iniciaremos creando el modulo de la conexión de la base de datos, el cual nombrare server.py y contendrá el siguiente código:
import os

from dotenv import load_dotenv
from sqlalchemy import create_engine

lee las variables de entorno

load_dotenv()

def create_engine_data()->"create_engine":
ENGINE_DB = os.getenv('ENGINE_DB')
CONNECTOR = os.getenv('CONNECTOR')
USER_DB = os.getenv('USER_DB')
PSW = os.getenv('PSW')
IP_SERVER = os.getenv('IP_SERVER')
DATABASE = os.getenv('DATABASE')
try:
engine = create_engine(
f'{ENGINE_DB}+{CONNECTOR}://{USER_DB}:\
{PSW}@{IP_SERVER}/{DATABASE}')
except Exception as ex:
raise ex
return engine

Importamos las variables de entorno con el uso de os.getenv, el nombre de cada constante es como se enuentra en nuestro archivo .env, luego procedemos a tratar de crear una instancia del motor de base de datos en el caso que la conexión se logre retornamos la instancia y si existe algún tipo de problema al momento de realizar la conexión nos levantaría una excepción.

Nota: Si están trabajando con un programa un poco mas complejo, lo adecuado seria importar el dontev y cargar los dotenv al inicio del programa y no en el archivo server.py quedando las importaciones con dos líneas menos.

import os

from sqlalchemy import create_engine

Fin de las importaciones

La función se mantiene intacta

Con el modulo de conexión al servidor de base de datos creado, podemos crear el modulo o los modelos de la base de datos que necesitaremos. Es importante aclarar que no es necesario modelar toda la base de datos, si solo necesitamos dos o tres tablas para trabajar basta con modelar estas tablas, asimismo no es necesario modelar toda la tabla si vamos a trabajar con 2 o 3 campos podemos modelar solamente esos campos.

Nuestra tabla usuarios tiene los siguientes campos:

A modo de ejemplo dejare por fuera el campo telephone, solamente porque no lo necesitaría.

El código de nuestro modelo de la tabla seria como el siguiente:
from sqlalchemy import Column
from sqlalchemy import String
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class Usuario(Base):
#Nombre de la tabla en la base de datos
tablename = "usuarios"

username = Column(String(30), primary_key=True)
name = Column(String(50))
surname = Column(String(50))
email = Column(String(60), unique=True)

#Representacion para el programador
def __repr__(self):
    return f"Usuario(username={self.username!r}, name={self.name!r},\
        surname={self.surname!r}, email={self.email!r})"

Con nuestro modelo hecho podemos continuar con el proceso de acceder a los datos en nuestra base de datos mediante el ORM, debemos crear la consulta o query con el ORM, en este caso hare dos consultas una que obtenga todos los usuarios y otra que obtenga los usuarios que cumplen con un criterio en especifico.

from sqlalchemy import select
from sqlalchemy.orm import Session

from server import create_engine_data
from models import Usuario

ENGINE = create_engine_data()
SESSION = Session(ENGINE)

def obtener_usuarios(name=''):
if name == '':
stmt = select(Usuario)
else:
stmt = select(Usuario).where(Usuario.name==name)
for usuario in SESSION.scalars(stmt):
print(usuario)

La consulta SQL para extraer todos lo usuarios seria
SELECT username, name, surname, telephone FROM usuarios

Y La segunda consulta SQL para extraer los usuarios que cumplen con un criterio seria
SELECT username, name, surname, telephone FROM usuario where name = name

Si deseamos ejecutar el código SQL de forma directa con ayuda de SQLAlchemy podemos hacerlo con ayuda de la instancia del servidor:

from server import create_engine_data
with ENGINE.connect() as conn:
data = conn.execute("SELECT username, name, surname, telephone FROM usuarios")

for usuario in data:
    print(usuario)

Nota: Debemos tener en cuenta que si migramos de motor de base de datos seria mas dificultoso el proceso de migración ya que deberíamos de verificar y reestructurar cada una de las sentencias.

SQLAlchemy Intro

Eduardo Santos — Wed, 19 Oct 2022 22:47:09 +0000

SQLAlchemy es una librería que facilita la comunicación entre programas hecho en Python y base de datos. Mayormente es utilizada como una ORM (Object Relational Mapper) que traduce las clases de Python a tablas en base de datos relacionales y que automáticamente convierte las funciones utilizadas a sentencias SQL. SQLAlchemy como la mayoría de herramientas provee una interfaz estándar que nos permite como desarrolladores crear la estructura y código de forma agnóstica del motor de base de datos a comunicarse, esto nos permite migrar de un motor a otro solo cambiando la cadena de conexión.

SQLAlchemy interactúa de forma directa con la API que integra la Base de datos de forma directa permitiéndonos utilizar funciones como connect, close, commit, rollback etc. Para cada motor de base de datos debemos de instalar el DBAPI que mas se adapte a nuestras necesidades para conectarnos a la base de datos.

Siempre que deseemos usar SQLAlchemy para interactuar con una base de datos, necesitamos crear una instancia de Engine (motor), esta instancia es la encargada de administrar dos factores cruciales Pools y Dialects.

Para crear la instancia del motor de base de datos utilizamos el siguiente código:

from sqlalchemy import create_engine

engine = create_engine(f’{motor_DB}://{user}:{pass}@{dirección_server}:{puerto}/{base_datos}’)

La estructura anterior tiene variables que pueden ser extraídas de un archivo .env para mejorar el proceso de desplegué del programa. En algunos casos será necesario pasarle el parámetro del driver o un conector ODBC.

Las conexiones agrupadas son la implementación mas utilizada en el patrón de grupos de objetos, estos grupos son utilizados como objetos pre inicializados en cache para que estén listos para su uso. En lugar de invertir tiempo creando objetos que son utilizados frecuentemente el programa utiliza un objeto preexistente. La principal razón para utilizar este patron de diseño es para mejorar el performance. Con las conexiones a base de datos el tiempo consumido y la cantidad de recursos mal utilizados volverían nuestra aplicación lenta y pesa para cualquier servidor.

SQLAlchemy Dialects nos permite crear aplicaciones que se comunican con diferentes motores de base de datos en la misma API. Los motores de base de datos mas populares han agregado el uso de SQL dentro de dialects, sin embargo, esto en algunas ocasiones puede generar un serio problema debido a que el código SQL no es un estándar entre los diferentes motores de base de datos, ya que estos últimos han personalizado la sintaxis para obtener los mismos resultados.

Para evitar que la migración de un motor a otro genere cambios significativos en nuestro código, se recomienda hacer uso del ORM como tal, crear la conexión de base de datos, crear los modelos de las tablas y los queries necesarios, con esto podremos asegurarnos que nuestra aplicación podrá mudarse de motor de base de datos con los cambios mínimos.

En el próximo post estaremos creando un ejemplo donde se incluyan el código SQL y una consulta con el ORM.

Consumir datos de SQL Server en Python

Eduardo Santos — Tue, 18 Oct 2022 03:06:54 +0000

En algún momento nos hemos encontrado con la necesidad de consumir datos o comunicarnos con una base de datos en SQL Server desde Django, a pesar que de forma nativa no tenemos soporte, te dejo dos opciones que me han ayudado a realizar dicha tarea. A pesar que existen diferentes herramientas es importante tener en cuenta que es lo que realmente necesitamos por ejemplo un ORM como SQLAlchemy o Peewee o un conector odbc como Pyodbc.

Primero si solo necesitas consumir datos mediante consulta sql o realizar pequeñas modificaciones te recomiendo Pyodbc, este conector no solo te funciona para SQL Server, puedes utilizarlo con MySql, Access, Oracle, Postgresql etc. Lograr conectar tu base de datos mediante ODBC es bastante fácil como veras a continuación:

import pyodbc
connection = pyodbc.connect(f'''{SU_DRIVER};SERVER={ip_server};port=1433;
                                   DATABASE={db_name};UID={user_name};
                                   PWD={password};Mars_Connection=Yes;
                                   TrustServerCertificate=yes;''')
cursor =  connection.cursor()

#Consulta para obtener todos los usuarios de la base de datos
consulta = f"""select * from users”””
cursor.execute(consulta)
data = cursor.fetchall()

En el segundo escenario necesitamos herramientas mas potentes y ahí es donde los ORMs como SQLAlchemy y Peewee nos ayudan en nuestra tarea de optimizar consultas, ya sea que necesitamos que nuestra aplicación consuma datos SQL Server o que la base de datos de nuestro proyecto se encuentre en dicho motor de base de datos, SQLAlchemy es el ORM que prefieren la gran mayoría de empresas para trabajar con datos no solamente SQL Server si no con MySql, Mariadb, Oracle, Postgresql, Firebird etc.

SQLAlchemy es un Mapeador Relacional de Objetos ORM (por sus siglas en ingles) esto nos permite crear clases como Producto y Orden y que ambos tengan una lista relacional de instancias, a diferencia de Pyodbc para aprovechar todo el poder de la herramienta debemos de crear clases y establecer las relaciones entre ellas, si existen atributos que no son necesarios para nosotros no es necesario definirlos en el modelo (de momento). El siguiente código es un ejemplo para definir la tabla Users y establecer una conexión con la base de datos.

from sqlalchemy import select
from sqlalchemy import create_engine
from sqlalchemy import Column
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy.orm import Session
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import relationship

#Linea de conexion
engine = create_engine( f'mssql+pyodbc://{user}:{password} @{ip_server}/{base_datos}?Driver={driver} ',
                            connect_args = {
                                            'TrustServerCertificate':'yes',
                                            'Mars_Connection':'yes'
                                            })

Base = declarative_base()

#Creamos la instancia de la sesión de nuestro servidor
session = Session(engine)

#Definimos la clase usuario con los atributos que necesitamos
class User(Base):
    __tablename__ = "users"

    user_name = Column(String(15), primary_key=True)
    name = Column(String(30))
    email = Column(String(100))

    #Representacion del objeto
    def __repr__(self):
        return f"User(user_name={self.user_name!r}, name={self.name!r}”

#Realiza la consulta sql 
stmt = select(User)

for user in session.scalars(stmt):
    print(user)

A pesar que el código tiene mas líneas que la versión de Pyodbc, con solo cambiar la línea de conexión a otro motor de base de datos nuestro código y consultas seguirían funcionando de la misma forma.
Estaré escribiendo mas sobre SQLAlchemy en los siguientes posts.

Optimizar sentencia if, elif, else

Eduardo Santos — Sun, 16 Oct 2022 20:57:53 +0000

En algún momento de nuestra vida como desarrolladores nos enfrentamos a controlar una situación con diferentes opciones, una de ellas podría ser controlar los días de la semana o los meses.
Para poder enfrentar esta situación escribimos un código como el siguiente:

if dia == 1:
    return “lunes”
elif dia == 2:
    return “martes”
…
elif dia == 7:
    return “domingo”
else:
    return “dia no controlado”

Pero podríamos eficientar este proceso haciendo uso de diccionarios, esto nos permitirá reducir significativamente las líneas de código, y generar un código mucho mas limpio.

DICT_DIAS = {
    '1': 'lunes',
    '2': 'martes',
    '3': 'miercoles',
    '4': 'jueves',
    '5': 'viernes',
    '6': 'sabado',
    '7': 'domingo'
}

def obtener_dia(valor:str)->str:
    return DICT_DIAS.get(valor,'Valor no controlado')

Con el código anterior no solamente logramos optimizar nuestro código si no que emula la función de la sentencia case, en caso que el valor buscado no exista devolvera 'Valor no controlado'