Un fichier de commande, trois services
Une cliente dépose un fichier de commande dans un bucket S3. Une minute plus tard, la commande est une ligne dans une table DynamoDB. Personne n’a appuyé sur un bouton entre les deux.
Trois petites étapes ont suffi : un événement, une file et une fonction. Cette leçon suit un dépôt à travers elles, et montre ce qui arrive quand une étape échoue ou tourne deux fois.
À la fin de cette leçon
Remarque
Je peux suivre un dépôt à travers une file et une fonction jusqu’à une table.
Mots-clés
| Mot | Ce que ça veut dire | Exemple |
|---|---|---|
| Chaîne de traitement | des étapes qui se passent le travail, l’une après l’autre | |
| Événement | une note qui dit que quelque chose s’est passé | |
| Message | un morceau de texte qui attend dans une file | |
| Élément | une ligne rangée dans une table DynamoDB | |
| Idempotent | sans danger à exécuter deux fois, pour le même résultat | écrire un élément seulement s’il est nouveau |
Les cellules que les étudiants remplissent restent vides dans cet aperçu.
Chaque étape passe du texte à la suivante
Le dépôt fait envoyer un événement par S3. Une règle le range dans une file SQS sous forme de message. Une fonction lit le message et écrit un élément dans une table.
Une file transporte du texte, pas des dictionnaires. L’événement devient donc une chaîne JSON, et la fonction la retransforme. Cet événement a la forme décrite par la documentation, pas celle d’une pile faite pour vous.
Exécutez : un événement devient le corps d’un message
import json
# La forme documentée d’un événement S3, réduite à ce que l’on lit.
evenement = {"Records": [{
"eventName": "ObjectCreated:Put",
"s3": {"bucket": {"name": "boutique-depots"},
"object": {"key": "commandes/ana/3.json"}}}]}
# La file transporte du texte : l’événement devient une chaîne JSON.
message = {"MessageId": "m-1", "ReceiptHandle": "h-1", "Body": json.dumps(evenement)}
print(type(message["Body"]).__name__)
# Le travailleur retransforme le texte en dictionnaire.
enregistrement = json.loads(message["Body"])["Records"][0]
print(enregistrement["s3"]["object"]["key"])
morceaux = enregistrement["s3"]["object"]["key"].split("/")
print(morceaux)
# Remplacez la clé par "commandes/ben/7.json" et exécutez de nouveau.
Un message peut arriver deux fois
Une file promet une livraison au moins une fois. De temps en temps, elle rend le même message, par exemple quand un travailleur a été lent à finir.
La fonction doit donc être idempotente. Écrire la commande une seconde fois ne doit pas en faire une deuxième. Une écriture conditionnelle y parvient : on écrit l’élément seulement si rien n’est rangé sous cette clé.
Erreur fréquente
Attention
Beaucoup croient que chaque message arrive exactement une fois. Ce n’est pas vrai : une file peut en livrer un plus d’une fois. Écrivez chaque gestionnaire pour qu’une répétition ne change rien. La seconde écriture n’est pas une erreur qui fait tout arrêter : elle dit que le travail est fait.
Exécutez : la seconde écriture est refusée
table = {}
def ecrire_une_fois(table, element):
cle = element["client"]["S"] + "#" + element["numero_commande"]["N"]
if cle in table:
return {"error": {"Code": "ConditionalCheckFailedException",
"HTTPStatusCode": 400}}
table[cle] = element
return {}
element = {"client": {"S": "ana"}, "numero_commande": {"N": "3"},
"statut": {"S": "recue"}}
premiere = ecrire_une_fois(table, element)
seconde = ecrire_une_fois(table, element)
print(premiere)
print(seconde["error"]["Code"])
print(len(table))
# Changez le numero_commande du second élément en "4". Combien d’éléments maintenant ?
Supprimer un message une fois le travail fait
Un travailleur prend un lot de messages. L’un d’eux est cassé. Les autres sont bons.
Traitez chaque message à part. Une fois le travail enregistré, supprimez ce message. Un message que vous n’avez pas pu traiter reste dans la file et revient après son délai de visibilité. Rien n’est perdu, et rien n’est supprimé trop tôt.
Exécutez : un lot avec un mauvais message
file = [
{"MessageId": "m-1", "Body": "commandes/ana/3.json"},
{"MessageId": "m-2", "Body": "casse"},
{"MessageId": "m-3", "Body": "commandes/ben/7.json"},
]
faits = []
restes = []
for message in file:
morceaux = message["Body"].split("/")
if len(morceaux) == 3:
faits.append(message["MessageId"])
else:
restes.append(message["MessageId"])
print("supprimés :", faits)
print("restés :", restes)
# Remplacez "casse" par "commandes/cai/9.json". Que reste-t-il maintenant ?