Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Programmation d’Amazon DynamoDB avec Python et Boto3

Ce guide de programmation fournit une orientation aux programmeurs qui souhaitent utiliser Amazon DynamoDB avec Python. Découvrez les différentes couches d’abstraction, la gestion de la configuration, la gestion des erreurs, le contrôle des politiques de nouvelles tentatives, la gestion de keep-alive, etc.

À propos de Boto

Vous pouvez accéder à DynamoDB depuis Python à l'aide du AWS SDK officiel pour Python, communément appelé Boto3. Le nom Boto (prononcé boh-toh) vient d’un dauphin d’eau douce originaire du fleuve Amazon. La bibliothèque Boto3 est la troisième version majeure de la bibliothèque, publiée pour la première fois en 2015. La bibliothèque Boto3 est assez volumineuse, car elle prend en charge tous les AWS services, pas seulement DynamoDB. Cette orientation cible uniquement les parties de Boto3 pertinentes pour DynamoDB.

Boto est maintenu et publié par AWS un projet open source hébergé sur. GitHub Il est divisé en deux packages : Botocore et Boto3.

Comme ces projets sont hébergés sur GitHub, vous pouvez consulter le code source, suivre les problèmes en cours ou soumettre vos propres problèmes.

Utilisation de la documentation Boto

Démarrez avec la documentation Boto grâce aux ressources suivantes :

Comprendre les couches d’abstraction du client et des ressources

Les deux interfaces que vous allez utiliser sont l’interface client et l’interface de ressources.

Voici un exemple d’insertion d’un élément à l’aide de l’interface client. Vous remarquerez que toutes les valeurs sont transmises sous forme de carte avec la clé indiquant leur type (« S » pour chaîne, « N » pour nombre) et leur valeur sous forme de chaîne. C’est ce que l’on appelle le format JSON DynamoDB.

import boto3

dynamodb = boto3.client('dynamodb')

dynamodb.put_item(
    TableName='YourTableName',
    Item={
        'pk': {'S': 'id#1'},
        'sk': {'S': 'cart#123'},
        'name': {'S': 'SomeName'},
        'inventory': {'N': '500'},
        # ... more attributes ...
    }
)

Voici la même opération PutItem à l’aide de l’interface de ressources. La saisie des données est implicite :

import boto3

dynamodb = boto3.resource('dynamodb')

table = dynamodb.Table('YourTableName')

table.put_item(
    Item={
        'pk': 'id#1',
        'sk': 'cart#123',
        'name': 'SomeName',
        'inventory': 500,
        # ... more attributes ...
    }
)

Si nécessaire, vous pouvez effectuer une conversion entre le JSON normal et le JSON DynamoDB à l’aide des classes TypeSerializer et TypeDeserializer fournies avec boto3 :

def dynamo_to_python(dynamo_object: dict) -> dict:
    deserializer = TypeDeserializer()
    return {
        k: deserializer.deserialize(v) 
        for k, v in dynamo_object.items()
    }  
  
def python_to_dynamo(python_object: dict) -> dict:
    serializer = TypeSerializer()
    return {
        k: serializer.serialize(v)
        for k, v in python_object.items()
    }

Voici comment effectuer une requête à l’aide de l’interface client. Elle exprime la requête sous forme de construction JSON. Elle utilise une chaîne KeyConditionExpression qui nécessite une substitution de variable pour gérer tout conflit de mots clés potentiel :

import boto3

client = boto3.client('dynamodb')

# Construct the query
response = client.query(
    TableName='YourTableName',
    KeyConditionExpression='pk = :pk_val AND begins_with(sk, :sk_val)',
    FilterExpression='#name = :name_val',
    ExpressionAttributeValues={
        ':pk_val': {'S': 'id#1'},
        ':sk_val': {'S': 'cart#'},
        ':name_val': {'S': 'SomeName'},
    },
    ExpressionAttributeNames={
        '#name': 'name',
    }
)

La même opération de requête utilisant l’interface de ressources peut être raccourcie et simplifiée :

import boto3
from boto3.dynamodb.conditions import Key, Attr

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('YourTableName')

response = table.query(
    KeyConditionExpression=Key('pk').eq('id#1') & Key('sk').begins_with('cart#'),
    FilterExpression=Attr('name').eq('SomeName')
)

Enfin, imaginez que vous souhaitiez obtenir la taille approximative d’une table (c’est-à-dire les métadonnées conservées sur la table et mises à jour toutes les 6 heures environ). Avec l’interface client, vous effectuez une opération describe_table() et vous extrayez la réponse de la structure JSON renvoyée :

import boto3

dynamodb = boto3.client('dynamodb')

response = dynamodb.describe_table(TableName='YourTableName')
size = response['Table']['TableSizeBytes']

Avec l’interface de ressources, la table exécute implicitement l’opération de description et présente les données directement sous forme d’attribut :

import boto3

dynamodb = boto3.resource('dynamodb')

table = dynamodb.Table('YourTableName')
size = table.table_size_bytes

Utilisation de la ressource de table batch_writer

Une commodité disponible uniquement avec la ressource de table de niveau supérieur est batch_writer. DynamoDB prend en charge les opérations d’écriture par lots, ce qui permet d’effectuer jusqu’à 25 opérations de saisie ou de suppression par requête réseau. Un tel traitement par lots améliore l’efficacité en minimisant les allers-retours sur le réseau.

Avec la bibliothèque client de bas niveau, vous utilisez l’opération client.batch_write_item() pour exécuter des lots. Vous devez diviser manuellement votre travail en lots de 25. Après chaque opération, vous devez également demander à recevoir une liste des éléments non traités (certaines opérations d'écriture peuvent réussir tandis que d'autres peuvent échouer). Vous devez ensuite retransmettre ces éléments non traités à une opération batch_write_item() ultérieure. Il y a une quantité importante de code standard.

La méthode Table.batch _writer crée un gestionnaire de contexte pour écrire des objets par lots. Elle présente une interface dans laquelle vous avez l’impression d’écrire des éléments un par un, mais en interne, elle les met en mémoire tampon et les envoie par lots. Elle gère également implicitement les nouvelles tentatives d’éléments non traités.

dynamodb = boto3.resource('dynamodb')

table = dynamodb.Table('YourTableName')

movies = # long list of movies in {'pk': 'val', 'sk': 'val', etc} format
with table.batch_writer() as writer:
    for movie in movies:
        writer.put_item(Item=movie)

Exemples supplémentaires de code qui explorent les couches client et ressources

Vous pouvez également consulter les référentiels d’exemples de code suivants qui explorent l’utilisation des différentes fonctions, en utilisant à la fois le client et les ressources :

Compréhension de la façon dont les objets Client et Ressource interagissent avec les sessions et les threads

Si vous ne créez pas votre propre session boto3.resource() et que vous boto3.client() utilisez une session partagée par défaut. C'est bien dans le code monothread, mais dans le code multithread, chaque thread crée des objets à partir de la même session, ce qui correspond au modèle non sécurisé décrit précédemment.

Partagez un client sur plusieurs sujets

Comme un client est compatible avec les threads, l'approche la plus simple et la plus efficace consiste à créer un seul client dans votre thread principal et à le remettre à chaque travailleur. Ils partagent tous le même pool de connexions.

Créé une fois, dans le fil principal, puis partagé avec tous les fils de discussion :

session = boto3.Session()
client = session.client("dynamodb")

Donnez à chaque fil sa propre ressource

Comme une ressource n'est pas adaptée aux threads, chaque thread a besoin de la sienne. Créez une nouvelle session dans le thread et créez la ressource à partir de celle-ci, afin que deux threads ne créent pas de ressources à partir de la même session à la fois.

S'exécute dans chaque thread :

session = boto3.Session()
dynamodb = session.resource("dynamodb")

Personnalisation de l’objet Configuration

Lorsque vous créez un objet Client ou Ressource, vous pouvez transmettre des paramètres nommés facultatifs pour personnaliser le comportement. Le paramètre nommé config déverrouille diverses fonctionnalités. Il s’agit d’une instance de botocore.client.Config et la documentation de référence de Configuration montre tout ce qu’il vous expose pour que vous puissiez le contrôler. Le Guide de Configuration fournit une bonne vue d’ensemble.

Configuration pour les délais d’expiration

L’une des utilisations d’une configuration personnalisée consiste à ajuster les comportements réseau :

Les délais d’expiration de 60 secondes sont excessifs pour DynamoDB. Cela signifie qu’un problème réseau passager retardera le client d’une minute avant qu’il ne puisse réessayer. Le code suivant réduit le délai d'attente à 10 secondes :

import boto3
from botocore.config import Config

my_config = Config(
   connect_timeout = 10.0,
   read_timeout = 10.0
)
dynamodb = boto3.resource('dynamodb', config=my_config)

Pour en savoir plus sur les délais d'expiration, consultez la section Réglage des paramètres de requête HTTP du SDK AWS Java pour les applications DynamoDB sensibles à la latence. Notez que le kit Java SDK possède plus de configurations de délai d’expiration que Python.

Configuration pour keep-alive

Si vous utilisez botocore 1.27.84 ou une version ultérieure, vous pouvez également contrôler le protocole TCP : Keep-Alive

La définition de TCP Keep-Alive sur True permet de réduire les latences moyennes. Voici un exemple de code qui définit conditionnellement TCP sur true lorsque vous disposez Keep-Alive de la bonne version de botocore :

import botocore
import boto3
from botocore.config import Config

required_version = "1.27.84"
current_version = botocore.__version__

my_config = Config(
   connect_timeout = 0.5,
   read_timeout = 0.5
)
if tuple(map(int, current_version.split("."))) >= tuple(map(int, required_version.split("."))):
    my_config = my_config.merge(Config(tcp_keepalive = True))

dynamodb = boto3.resource('dynamodb', config=my_config)

Il y a une limite à la durée pendant laquelle une connexion inactive peut être maintenue en vie. Envisagez d’envoyer des demandes périodiques (disons toutes les minutes) si vous avez une connexion inactive mais que vous souhaitez que la prochaine demande utilise une connexion déjà établie.

Configuration pour les nouvelles tentatives

La configuration accepte également un dictionnaire appelé nouvelles tentatives dans lequel vous pouvez spécifier le comportement de nouvelle tentative souhaité. Les nouvelles tentatives se produisent dans le kit SDK lorsque celui-ci reçoit une erreur et que l’erreur est de type passager. Si une erreur fait l’objet d’une nouvelle tentative en interne (et qu’une nouvelle tentative produit finalement une réponse de réussite), aucune erreur n’est détectée du point de vue du code appelant, juste une latence légèrement élevée. Voici les valeurs que vous pouvez spécifier :

Vous trouverez une définition détaillée de ces modes de nouvelle tentative dans le Guide des nouvelles tentatives ainsi que dans Comportement des tentatives dans la référence du kit SDK.

Voici un exemple qui utilise explicitement la politique de nouvelles tentatives legacy avec un maximum de 3 demandes au total (2 nouvelles tentatives) :

import boto3
from botocore.config import Config

my_config = Config(
   connect_timeout = 1.0,
   read_timeout = 1.0,
   retries = {
     'mode': 'legacy',
     'total_max_attempts': 3
   }
)
dynamodb = boto3.resource('dynamodb', config=my_config)

DynamoDB étant un système à haute disponibilité et à faible latence, vous souhaiterez peut-être être plus agressif en ce qui concerne la vitesse des nouvelles tentatives que ne le permettent les politiques de nouvelles tentatives intégrées. Vous pouvez implémenter votre propre politique de nouvelles tentatives en fixant le nombre maximum de tentatives à 0, en détectant vous-même les exceptions et en réessayant le cas échéant à partir de votre propre code au lieu de compter sur boto3 pour effectuer des nouvelles tentatives implicites.

Si vous gérez votre propre stratégie de nouvelle tentative, vous devez faire la différence entre les limitations et les erreurs :

Pour plus d'informations sur les modifications à venir du comportement de nouvelle tentative par défaut dans tous les AWS SDK, voir Annonce d'un comportement de nouvelle tentative mis à jour pour les AWS SDK et les outils sur le AWS blog Developer Tools. Pour plus d'informations sur les modifications spécifiques àPython, consultez la discussion sur la mise à jour du comportement des nouvelles tentatives (#4789) sur le GitHub site Web.

Configuration pour le nombre maximum de connexions au groupe

Enfin, la configuration vous permet de contrôler la taille du groupe de connexions :

Cette option contrôle le nombre maximum de connexions HTTP à conserver en groupe en vue de leur réutilisation. Un pool différent est conservé pour chaque client. Si un thread d'un client souhaite établir une connexion et que le pool est vide, il établira une nouvelle connexion. Si un thread essaie de renvoyer une connexion au pool après utilisation et trouve que le pool est plein, il ferme la connexion. Si vous pensez que plus de 10 threads seront envoyés à un seul client, pensez à augmenter la taille du pool pour qu'elle corresponde au nombre de threads.

import boto3
from botocore.config import Config

my_config = Config(
   max_pool_connections = 20
)

session = boto3.Session()

# One client with a larger connection pool
client = session.client("dynamodb", config=my_config)

Chaque ressource possède également un pool de connexions, mais un objet Resource n'est pas sécurisé pour les threads, donc n'en partagez pas un entre les threads. Donnez à chaque fil sa propre ressource.

Gestion des erreurs

AWS les exceptions de service ne sont pas toutes définies statiquement dans Boto3. Cela est dû au fait que les erreurs et les exceptions relatives AWS aux services varient considérablement et sont susceptibles de changer. Boto3 enveloppe toutes les exceptions de service sous forme de ClientError et expose les détails en tant que JSON structuré. Par exemple, une réponse d’erreur peut être structurée comme suit :

{
    'Error': {
        'Code': 'SomeServiceException',
        'Message': 'Details/context around the exception or error'
    },
    'ResponseMetadata': {
        'RequestId': '1234567890ABCDEF',
        'HostId': 'host ID data will appear here as a hash',
        'HTTPStatusCode': 400,
        'HTTPHeaders': {'header metadata key/values will appear here'},
        'RetryAttempts': 0
    }
}

Le code suivant détecte toutes les exceptions ClientError et examine la valeur de chaîne de caractères du Code dans Error pour déterminer l’action à entreprendre :

import botocore
import boto3

dynamodb = boto3.client('dynamodb')

try:
    response = dynamodb.put_item(...)

except botocore.exceptions.ClientError as err:
    print('Error Code: {}'.format(err.response['Error']['Code']))
    print('Error Message: {}'.format(err.response['Error']['Message']))
    print('Http Code: {}'.format(err.response['ResponseMetadata']['HTTPStatusCode']))
    print('Request ID: {}'.format(err.response['ResponseMetadata']['RequestId']))

    if err.response['Error']['Code'] in ('ProvisionedThroughputExceededException', 'ThrottlingException'):
        print("Received a throttle")
    elif err.response['Error']['Code'] == 'InternalServerError':
        print("Received a server error")
    else:
        raise err

Certains codes d’exception (mais pas tous) ont été matérialisés sous forme de classes de premier niveau. Vous pouvez choisir de les gérer directement. Lorsque vous utilisez l’interface Client, ces exceptions sont renseignées dynamiquement sur le client et vous pouvez les intercepter à l’aide de l’instance client, comme suit :

except ddb_client.exceptions.ProvisionedThroughputExceededException:

Lorsque vous utilisez l’interface Ressource, vous devez utiliser .meta.client pour passer de la ressource au Client sous-jacent pour accéder aux exceptions, comme ceci :

except ddb_resource.meta.client.exceptions.ProvisionedThroughputExceededException:

Pour consulter la liste des types d’exception matérialisés, vous pouvez générer la liste de manière dynamique :

ddb = boto3.client("dynamodb")
print([e for e in dir(ddb.exceptions) if e.endswith('Exception') or e.endswith('Error')])

Lorsque vous effectuez une opération d’écriture avec une expression de condition, vous pouvez demander qu’en cas d’échec de l’expression, la valeur de l’élément soit renvoyée dans la réponse d’erreur.

try:
    response = table.put_item(
        Item=item,
        ConditionExpression='attribute_not_exists(pk)',
        ReturnValuesOnConditionCheckFailure='ALL_OLD'
    )
except table.meta.client.exceptions.ConditionalCheckFailedException as e:
    print('Item already exists:', e.response['Item'])

Pour en savoir plus sur la gestion des erreurs et les exceptions, consultez :

Logging

La bibliothèque boto3 s’intègre au module de journalisation intégré de Python pour suivre ce qui se passe pendant une session. Pour contrôler les niveaux de journalisation, vous pouvez configurer le module de journalisation :

import logging

logging.basicConfig(level=logging.INFO)

Cela configure l’enregistreur racine pour qu’il journalise INFO et les messages de niveau supérieur. Les messages de journalisation moins graves que le niveau sont ignorés. Les niveaux de journalisation peuvent être DEBUG, INFO, WARNING, ERROR et CRITICAL. La valeur par défaut est WARNING.

Les enregistreurs de boto3 sont hiérarchiques. La bibliothèque utilise plusieurs enregistreurs différents, chacun correspondant à différentes parties de la bibliothèque. Vous pouvez contrôler séparément le comportement de chacun :

D’autres bibliothèques utilisent également les journaux. En interne, boto3 utilise le fichier tiers urllib3 pour la gestion des connexions HTTP. Lorsque la latence est importante, vous pouvez consulter ses journaux pour vous assurer que votre pool est bien utilisé en regardant quand urllib3 établit une nouvelle connexion ou ferme une connexion inactive.

L’extrait de code suivant définit la plupart des connexions sur la journalisation dans INFO avec la journalisation DEBUG pour l’activité du point de terminaison et du groupe de connexions :

import logging

logging.getLogger('boto3').setLevel(logging.INFO)
logging.getLogger('botocore').setLevel(logging.INFO)
logging.getLogger('botocore.endpoint').setLevel(logging.DEBUG)
logging.getLogger('urllib3.connectionpool').setLevel(logging.DEBUG)

Hooks d’événement

Botocore émet des événements au cours des différentes étapes de son exécution. Vous pouvez enregistrer des gestionnaires pour ces événements afin que chaque fois qu’un événement est émis, votre gestionnaire soit appelé. Cela vous permet d’étendre le comportement du botocore sans avoir à modifier les composants internes.

Supposons, par exemple, que vous souhaitiez suivre chaque fois qu’une opération PutItem est appelée sur une table DynamoDB de votre application. Vous pouvez vous enregistrer sur l’événement 'provide-client-params.dynamodb.PutItem' à capturer et journaliser chaque fois qu’une opération PutItem est invoquée dans la session associée. Voici un exemple :

import boto3
import botocore
import logging

def log_put_params(params, **kwargs):
    if 'TableName' in params and 'Item' in params:
        logging.info(f"PutItem on table {params['TableName']}: {params['Item']}")

logging.basicConfig(level=logging.INFO)

session = boto3.Session()
event_system = session.events

# Register our interest in hooking in when the parameters are provided to PutItem
event_system.register('provide-client-params.dynamodb.PutItem', log_put_params)

# Now, every time you use this session to put an item in DynamoDB,
# it will log the table name and item data.
dynamodb = session.resource('dynamodb')
table = dynamodb.Table('YourTableName')
table.put_item(
    Item={
        'pk': '123',
        'sk': 'cart#123',
        'item_data': 'YourItemData',
        # ... more attributes ...
    }
)

Dans le gestionnaire, vous pouvez même manipuler les paramètres par programmation pour modifier le comportement :

params['TableName'] = "NewTableName"

Pour plus d’informations sur les événements, consultez botocore documentation on events et boto3 documentation on events.

Pagination et paginateur

Certaines demandes, telles que Query et Scan, limitent la taille des données renvoyées lors d’une seule demande et nécessitent que vous fassiez des demandes répétées pour extraire les pages suivantes.

Vous pouvez contrôler le nombre maximum d’éléments à lire pour chaque page à l’aide du paramètre limit. Par exemple, si vous voulez extraire uniquement les 10 derniers éléments, vous pouvez utiliser le paramètre limit. Notez que la limite correspond à la quantité qui doit être lue dans la table avant qu’un filtrage ne soit appliqué. Il n’est pas possible de spécifier que vous voulez exactement 10 éléments après le filtrage ; vous ne pouvez contrôler le nombre préfiltré et vérifier côté client que lorsque vous en avez réellement extrait 10. Quelle que soit la limite, chaque réponse a toujours une taille maximale de 1 Mo.

Si la réponse inclut une LastEvaluatedKey, cela indique qu’elle s’est terminée parce qu’elle a atteint une limite de nombre ou de taille. La clé est la dernière clé évaluée pour la réponse. Vous pouvez extraire cette LastEvaluatedKey et la transmettre à un appel de suivi en tant que ExclusiveStartKey afin de lire le fragment suivant à partir de ce point de départ. Si aucune LastEvaluatedKey n’est renvoyée, cela signifie qu’il n’y a plus d’éléments correspondant à Query ou Scan.

Voici un exemple simple (utilisant l’interface Ressource, mais l’interface Client suit le même schéma) qui lit au maximum 100 éléments par page et boucle jusqu’à ce que tous les éléments aient été lus.

import boto3
from boto3.dynamodb.conditions import Key

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('YourTableName')

query_params = {
    'KeyConditionExpression': Key('pk').eq('123') & Key('sk').gt(1000),
    'Limit': 100
}

while True:
    response = table.query(**query_params)

    # Process the items however you like
    for item in response['Items']:
        print(item)

    # No LastEvaluatedKey means no more items to retrieve
    if 'LastEvaluatedKey' not in response:
        break

    # If there are possibly more items, update the start key for the next page
    query_params['ExclusiveStartKey'] = response['LastEvaluatedKey']

Pour plus de commodité, boto3 peut le faire pour vous avec des paginateurs. Cependant, cela ne fonctionne qu’avec l’interface Client. Voici le code réécrit pour utiliser les paginateurs :

import boto3

dynamodb = boto3.client('dynamodb')

paginator = dynamodb.get_paginator('query')

query_params = {
    'TableName': 'YourTableName',
    'KeyConditionExpression': 'pk = :pk_val AND sk > :sk_val',
    'ExpressionAttributeValues': {
        ':pk_val': {'S': '123'},
        ':sk_val': {'N': '1000'},
    },
    'Limit': 100
}

page_iterator = paginator.paginate(**query_params)

for page in page_iterator:
    # Process the items however you like
    for item in page['Items']:
        print(item)

Pour plus d'informations, consultez le Guide des paginateurs et la référence de l'API pour. DynamoDB.Paginator.Query

Programmes d’attente

Les programmes d’attente offrent la possibilité d’attendre que quelque chose soit terminé avant de continuer. À l’heure actuelle, ils ne prennent en charge que l’attente de la création ou de la suppression d’une table. En arrière-plan, le programme d’attente effectue une vérification pour vous toutes les 20 secondes, jusqu’à 25 fois. Vous pouvez le faire vous-même mais l’utilisation d’un programme d’attente est élégante lors de l’automatisation de l’écriture.

Ce code montre comment attendre qu’une table particulière ait été créée :

# Create a table, wait until it exists, and print its ARN
response = client.create_table(...)
waiter = client.get_waiter('table_exists')
waiter.wait(TableName='YourTableName')
print('Table created:', response['TableDescription']['TableArn'])

Pour plus d’informations, consultez le Guide to Waiters et la Reference on Waiters.