Un archivo de pedido, tres servicios
Una clienta sube el archivo de su pedido a un bucket de S3. Un minuto después el pedido es una fila de una tabla de DynamoDB. Nadie ha pulsado ningún botón entre medias.
Lo hacen tres pasos pequeños: un evento, una cola y una función. Esta lección sigue una subida por ellos y muestra qué pasa si un paso falla o se ejecuta dos veces.
Al final de esta lección
Nota
Puedo seguir una subida por una cola y una función hasta una tabla.
Palabras clave
| Palabra | Qué significa | Ejemplo |
|---|---|---|
| Cadena | pasos que se pasan el trabajo uno al siguiente | |
| Evento | un aviso de que algo ha ocurrido | |
| Mensaje | un texto que espera en una cola | |
| Elemento | una fila guardada en una tabla de DynamoDB | item |
| Idempotente | seguro de ejecutar dos veces con el mismo resultado | guardar un elemento solo si es nuevo |
Las celdas que rellenan los estudiantes quedan en blanco en esta vista previa.
Cada paso pasa un texto al siguiente
La subida hace que S3 envíe un evento. Una regla pone ese evento en una cola de SQS como mensaje. Una función lee el mensaje y escribe un elemento en una tabla.
La cola lleva texto, no diccionarios. Por eso el evento se convierte en un texto JSON y la función lo convierte de nuevo. Este evento es la forma documentada, no uno que una pila haya hecho para ti.
Run: un evento pasa a cuerpo de mensaje
import json
# La forma documentada de un evento de S3, recortada a lo que leemos.
event = {"Records": [{
"eventName": "ObjectCreated:Put",
"s3": {"bucket": {"name": "tienda-subidas"},
"object": {"key": "pedidos/lucia/3.json"}}}]}
# La cola lleva texto, así que el evento se convierte en un texto JSON.
message = {"MessageId": "m-1", "ReceiptHandle": "h-1", "Body": json.dumps(event)}
print(type(message["Body"]).__name__)
# El trabajador convierte el texto de nuevo en un diccionario.
record = json.loads(message["Body"])["Records"][0]
print(record["s3"]["object"]["key"])
parts = record["s3"]["object"]["key"].split("/")
print(parts)
# Cambia la clave a "pedidos/marta/7.json" y vuelve a ejecutarlo.
Un mensaje puede llegar dos veces
Una cola promete entrega al menos una vez. De vez en cuando entrega otra vez el mismo mensaje, por ejemplo si un trabajador tardó en terminar.
Por eso la función debe ser idempotente. Escribir el pedido otra vez no debe crear un segundo pedido. Lo consigue una escritura condicional: guarda el elemento solo si no hay nada guardado con esa clave.
Un error frecuente
Advertencia
Mucha gente cree que cada mensaje llega una sola vez. No es así: una cola puede entregar un mensaje más de una vez. Escribe cada handler de modo que una repetición no cambie nada. La segunda escritura no es un error que deba romper el programa: significa que el trabajo ya está hecho.
Run: la segunda escritura se rechaza
table = {}
def put_item_once(table, item):
key = item["customer"]["S"] + "#" + item["order_id"]["N"]
if key in table:
return {"error": {"Code": "ConditionalCheckFailedException",
"HTTPStatusCode": 400}}
table[key] = item
return {}
item = {"customer": {"S": "lucia"}, "order_id": {"N": "3"},
"status": {"S": "received"}}
first = put_item_once(table, item)
second = put_item_once(table, item)
print(first)
print(second["error"]["Code"])
print(len(table))
# Crea otro elemento con order_id "4" y guárdalo en la segunda llamada. ¿Cuántos elementos hay ahora?
Borra un mensaje cuando su trabajo está hecho
Un trabajador toma un lote de mensajes. Uno está estropeado. Los demás están bien.
Atiende cada mensaje por separado. Cuando el trabajo está guardado, borra ese mensaje. Uno que no pudiste atender se queda en la cola y vuelve cuando termina su tiempo de visibilidad. No se pierde nada y nada se borra demasiado pronto.
Run: un lote con un mensaje malo
queue = [
{"MessageId": "m-1", "Body": "pedidos/lucia/3.json"},
{"MessageId": "m-2", "Body": "roto"},
{"MessageId": "m-3", "Body": "pedidos/marta/7.json"},
]
done = []
waiting = []
for message in queue:
parts = message["Body"].split("/")
if len(parts) == 3:
done.append(message["MessageId"])
else:
waiting.append(message["MessageId"])
print("borrados:", done)
print("quedan:", waiting)
# Cambia "roto" por "pedidos/pablo/9.json". ¿Qué queda ahora?