Producir y consumir eventos: tu primer flujo event-driven

1import boto3
2import json
3import random
4import time
5from datetime import datetime
6
7eb = boto3.client("events", endpoint_url="http://localhost:4566",
8                  region_name="eu-west-1", aws_access_key_id="test",
9                  aws_secret_access_key="test")
10
11productos = ["Laptop", "Teclado", "Monitor", "Auriculares", "Webcam", "SSD 1TB"]
12clientes = [f"CLI-{i}" for i in range(1, 20)]
13
14def generar_pedido():
15    """Genera un evento de pedido aleatorio."""
16    items = [{"producto": p, "precio": round(random.uniform(20, 500), 2)}
17             for p in random.sample(productos, k=random.randint(1, 3))]
18    total = round(sum(i["precio"] for i in items), 2)
19    return {
20        "Source": "servicio-pedidos",
21        "DetailType": "PedidoCreado",
22        "Detail": json.dumps({
23            "pedido_id": f"ORD-{random.randint(1000, 9999)}",
24            "cliente_id": random.choice(clientes),
25            "items": items,
26            "total": total,
27            "timestamp": datetime.utcnow().isoformat() + "Z",
28        }),
29        "EventBusName": "ecommerce-bus",
30    }
31
32# Producir 5 pedidos con 2s de separación
33for i in range(5):
34    evento = generar_pedido()
35    resp = eb.put_events(Entries=[evento])
36    detail = json.loads(evento["Detail"])
37    print(f"[{i+1}] Pedido {detail['pedido_id']} - Total: {detail['total']}EUR")
38    time.sleep(2)
39
40print("\nProductor finalizado. 5 eventos publicados.")

Un productor que simula pedidos aleatorios cada 2 segundos

1import boto3
2import json
3import time
4
5sqs = boto3.client("sqs", endpoint_url="http://localhost:4566",
6                   region_name="eu-west-1", aws_access_key_id="test",
7                   aws_secret_access_key="test")
8
9def consumir(queue_name: str, handler_name: str):
10    """Consume mensajes de una cola indefinidamente."""
11    queue_url = f"http://sqs.eu-west-1.localhost.localstack.cloud:4566/000000000000/{queue_name}"
12    print(f"[{handler_name}] Escuchando cola: {queue_name}...")
13    
14    while True:
15        resp = sqs.receive_message(
16            QueueUrl=queue_url,
17            MaxNumberOfMessages=5,
18            WaitTimeSeconds=20,
19        )
20        for msg in resp.get("Messages", []):
21            body = json.loads(msg["Body"])
22            detail = body.get("detail", body)
23            print(f"  [{handler_name}] Procesando pedido {detail.get('pedido_id', '?')}")
24            # Simular procesamiento
25            time.sleep(0.5)
26            sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=msg["ReceiptHandle"])
27            print(f"  [{handler_name}] OK - confirmado")
28
29# Ejecutar: python consumidor.py inventario
30# Ejecutar: python consumidor.py emails
31# Ejecutar: python consumidor.py almacen
32if __name__ == "__main__":
33    import sys
34    cola = sys.argv[1] if len(sys.argv) > 1 else "pedidos-nuevos"
35    consumir(cola, cola.upper())

Consumidor genérico que escucha una cola SQS con long polling

1# Patrón Request-Reply con correlation_id
2import uuid
3
4# El productor incluye un correlation_id
5evento_request = {
6    "Source": "servicio-checkout",
7    "DetailType": "VerificarPago",
8    "Detail": json.dumps({
9        "correlation_id": str(uuid.uuid4()),
10        "pedido_id": "ORD-5555",
11        "monto": 299.99,
12        "tarjeta_hash": "xxx-1234"
13    }),
14    "EventBusName": "ecommerce-bus",
15}
16
17# El consumidor (servicio-pagos) responde con OTRO evento
18evento_reply = {
19    "Source": "servicio-pagos",
20    "DetailType": "PagoVerificado",
21    "Detail": json.dumps({
22        "correlation_id": "mismo-uuid-del-request",
23        "pedido_id": "ORD-5555",
24        "aprobado": True,
25        "referencia_banco": "BNK-789456"
26    }),
27    "EventBusName": "ecommerce-bus",
28}

Request-reply: correlacionar petición y respuesta a través de eventos

1# Test unitario del handler (sin infraestructura)
2
3def handler_inventario(evento: dict) -> dict:
4    """Procesa un evento de pedido actualizando inventario."""
5    items = evento["detail"]["items"]
6    resultados = []
7    for item in items:
8        resultados.append({"producto": item["producto"], "stock_restado": item.get("cantidad", 1)})
9    return {"status": "ok", "actualizaciones": resultados}
10
11def test_handler_inventario():
12    evento = {
13        "detail-type": "PedidoCreado",
14        "detail": {
15            "pedido_id": "ORD-TEST",
16            "items": [
17                {"producto": "Laptop", "cantidad": 1},
18                {"producto": "Mouse", "cantidad": 2},
19            ]
20        }
21    }
22    resultado = handler_inventario(evento)
23    assert resultado["status"] == "ok"
24    assert len(resultado["actualizaciones"]) == 2
25    assert resultado["actualizaciones"][0]["stock_restado"] == 1

Testear handlers de eventos como funciones puras, sin infraestructura