Hace un tiempo me tocó integrar DynamoDB con Lambda en un proyecto pequeño: un backend serverless para registrar eventos de usuario. Nada del otro mundo en papel, pero me demoré un poco más de lo esperado por eso esta guía cubre desde qué es DynamoDB hasta cómo exponerlo con API Gateway, con una sección de debugging real, optimización de costos, y un FAQ con las preguntas que más se repiten en el grupo de la comunidad.
¿Qué es DynamoDB y cuándo usarlo?
DynamoDB es el servicio de base de datos NoSQL administrado de AWS. "Administrado" significa que no hay servidor que configurar, parches que aplicar ni réplicas que coordinar. Tú defines la tabla, insertas datos, y AWS gestiona el resto: replicación multi-AZ, backups, escalabilidad.
A diferencia de una base relacional, en DynamoDB no defines un esquema estricto con columnas fijas. Cada ítem puede tener atributos distintos. Lo que sí es obligatorio al crear la tabla es definir la clave primaria, que puede ser:
- Partition Key (PK) — clave simple, identifica de forma única cada ítem.
- Partition Key + Sort Key (PK + SK) — clave compuesta, útil para agrupar múltiples ítems bajo una misma partición (ej: todos los pedidos de un usuario).
Modelo mental correcto
El diseño en DynamoDB gira alrededor del patrón de acceso, no del modelo de datos. Primero define cómo vas a consultar, luego diseña la tabla. Es al revés que en SQL, y eso confunde al principio.
¿Cuándo tiene sentido? Cuando el patrón de acceso es predecible, necesitas escala sin fricción, y especialmente cuando estás construyendo arquitecturas serverless. Ambos son stateless, ambos escalan automáticamente, y ninguno requiere conexiones persistentes.
⚠️ Cuándo NO usarlo
Si necesitas JOINs complejos, consultas ad-hoc frecuentes sobre múltiples atributos, o transacciones ACID complejas entre varias entidades, considera RDS o Aurora. DynamoDB brilla en acceso de clave-valor y patrones bien definidos, no en exploración libre de datos.
El flujo que vamos a construir
Vamos a construir dos versiones: la primera es solo Lambda + DynamoDB (ideal para aprender los conceptos sin ruido). La segunda agrega API Gateway para tener un endpoint HTTP real.
Paso a paso: desde cero hasta funcional
1. Crear la tabla en DynamoDB
Desde la consola de AWS → DynamoDB → Create table. Configuración mínima:
- Table name: usuarios
- Partition key: userId — tipo String
- Capacity mode: On-demand (para pruebas y carga impredecible)
On-demand vs Provisionado
On-demand: pagas por cada unidad de lectura/escritura. Ideal para tráfico variable o proyectos nuevos donde no conoces el volumen.
Provisionado: defines RCUs y WCUs fijas. Más barato a escala predecible, pero si te quedas corto pagas con throttling.
2. Crear el rol IAM para Lambda
Este paso define qué puede hacer tu función Lambda. La regla de oro es least privilege: solo los permisos necesarios, nada más.
{
"Version": "2012-10-17",
"Statement": [{
"Sid": "DynamoDBUsuariosAccess",
"Effect": "Allow",
"Action": [
"dynamodb:PutItem",
"dynamodb:GetItem",
"dynamodb:UpdateItem",
"dynamodb:DeleteItem",
"dynamodb:Query"
],
// ARN específico de tu tabla — OJO, NO uses "*"
"Resource": "arn:aws:dynamodb:us-east-1:TU_ACCOUNT_ID:table/usuarios"
}]
}
Pilas con este error clásico de seguridad
No uses "Resource": "*" en producción. El ARN lo encuentras en la consola de DynamoDB en la pestaña "Additional info" de la tabla. Cópialo desde ahí, no lo construyas a mano.
función Lambda — CRUD completo
Lambda → Create function → Python 3.12, asigna el rol del paso anterior. El código cubre las cuatro operaciones básicas con manejo de errores real:
import json
import os
import logging
import boto3
from botocore.exceptions import ClientError
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
# --- Inicializar fuera del handler ---
# Lambda reutiliza el entorno entre invocaciones calientes (warm start).
# Inicializar boto3 una sola vez evita el overhead de reconexión en cada
# llamada. En producción con alta concurrencia, esto reduce latencia real.
dynamodb = boto3.resource('dynamodb')
TABLE_NAME = os.environ.get('TABLE_NAME', 'usuarios') # variable de entorno
table = dynamodb.Table(TABLE_NAME)
def lambda_handler(event, context):
"""
Dispatcher principal. Acepta invocación directa (tests) y via API Gateway.
El campo 'accion' determina qué función se ejecuta.
"""
# Compatibilidad con API Gateway (body llega como string JSON)
if 'httpMethod' in event or 'requestContext' in event:
try:
payload = json.loads(event.get('body') or '{}')
except json.JSONDecodeError:
return _resp(400, {'error': 'Body no es JSON válido'})
else:
payload = event # invocación directa desde consola o test
accion = payload.get('accion')
logger.info("Acción: %s", accion)
handlers = {
'crear': crear_usuario,
'obtener': obtener_usuario,
'actualizar': actualizar_usuario,
'eliminar': eliminar_usuario,
}
fn = handlers.get(accion)
if not fn:
return _resp(400, {
'error': f'Acción "{accion}" no reconocida',
'acciones_validas': list(handlers.keys())
})
try:
return fn(payload)
except ClientError as e:
code = e.response['Error']['Code']
logger.error("ClientError [%s]: %s", code, e)
return _resp(500, {'error': f'Error AWS: {code}'})
except Exception as e:
logger.error("Error inesperado", exc_info=True)
return _resp(500, {'error': 'Error interno del servidor'})
# --- CREAR ---
def crear_usuario(payload):
user_id = payload.get('userId')
nombre = payload.get('nombre')
if not user_id or not nombre:
return _resp(400, {'error': 'userId y nombre son campos requeridos'})
item = {'userId': user_id, 'nombre': nombre}
for campo in ['email', 'rol', 'activo']:
if payload.get(campo) is not None:
item[campo] = payload[campo]
try:
# attribute_not_exists(userId) lanza ConditionalCheckFailedException
# si el ítem ya existe. Así evitamos sobrescrituras silenciosas.
table.put_item(
Item=item,
ConditionExpression='attribute_not_exists(userId)'
)
except ClientError as e:
if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
return _resp(409, {'error': f'El userId "{user_id}" ya existe'})
raise # re-lanza si es otro ClientError
logger.info("Usuario creado: %s", user_id)
return _resp(201, {'mensaje': f'Usuario {user_id} creado', 'item': item})
# --- OBTENER ---
def obtener_usuario(payload):
user_id = payload.get('userId')
if not user_id:
return _resp(400, {'error': 'userId es requerido'})
response = table.get_item(
Key={'userId': user_id},
# ProjectionExpression: solo los atributos necesarios.
# Menos bytes transferidos = menos RCUs consumidas = menor costo.
ProjectionExpression='userId, nombre, email, rol, activo'
)
usuario = response.get('Item')
if not usuario:
return _resp(404, {'error': f'Usuario "{user_id}" no encontrado'})
return _resp(200, usuario)
# --- ACTUALIZAR ---
def actualizar_usuario(payload):
user_id = payload.get('userId')
# Extraer campos a actualizar, excluyendo metadatos del request
campos = {
k: v for k, v in payload.items()
if k not in ('accion', 'userId')
}
if not user_id or not campos:
return _resp(400, {'error': 'userId y al menos un campo son requeridos'})
# ExpressionAttributeNames escapa palabras reservadas de DynamoDB
# (name, status, role, count, etc.). Siempre usarlo en UpdateExpression.
update_expr = 'SET ' + ', '.join([f'#{k} = :{k}' for k in campos])
attr_names = {f'#{k}': k for k in campos}
attr_values = {f':{k}': v for k, v in campos.items()}
try:
table.update_item(
Key={'userId': user_id},
UpdateExpression=update_expr,
ExpressionAttributeNames=attr_names,
ExpressionAttributeValues=attr_values,
# Falla si el usuario no existe — mejor que actualizar nada silenciosamente
ConditionExpression='attribute_exists(userId)'
)
except ClientError as e:
if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
return _resp(404, {'error': f'Usuario "{user_id}" no existe'})
raise
return _resp(200, {'mensaje': f'Usuario {user_id} actualizado', 'campos': list(campos.keys())})
# --- ELIMINAR ---
def eliminar_usuario(payload):
user_id = payload.get('userId')
if not user_id:
return _resp(400, {'error': 'userId es requerido'})
# delete_item no lanza error si el ítem no existe — es idempotente por diseño
table.delete_item(Key={'userId': user_id})
return _resp(200, {'mensaje': f'Usuario {user_id} eliminado'})
# --- HELPER ---
def _resp(status_code: int, body: dict) -> dict:
"""Formato de respuesta compatible con API Gateway Lambda Proxy."""
return {
'statusCode': status_code,
'headers': {
'Content-Type': 'application/json',
'Access-Control-Allow-Origin': '*' # ajustar a dominio específico en prod
},
'body': json.dumps(body, ensure_ascii=False, default=str)
}
Probar la función — requests y respuestas reales
Lambda → pestaña Test → Create new event. Acá va cada operación con su request y la respuesta esperada. Si algo falla, la sección de debugging más abajo tiene el flujo de diagnóstico.
{
"accion": "crear",
"userId": "usr-001",
"nombre": "Ana Suárez",
"email": "ana@ejemplo.com",
"rol": "admin",
"activo": true
}
Top comments (0)