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 en utilisant le kit 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 services AWS, pas seulement DynamoDB. Cette orientation cible uniquement les parties de Boto3 pertinentes pour DynamoDB.

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

Ces projets étant hébergés sur GitHub, vous pouvez consulter le code source, suivre les problèmes en suspens 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 dans un lot. 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

L’objet Resource n’est pas adapté aux threads et ne doit pas être partagé entre les threads ou les processus. Pour plus d’informations, reportez-vous au Guide sur Ressource.

L’objet Client, en revanche, est généralement adapté aux threads, à l’exception de certaines fonctionnalités avancées. Pour plus d’informations, reportez-vous au Guide sur Clients.

L’objet Session n’est pas adapté aux threads. Ainsi, chaque fois que vous créez un objet Client ou Ressource dans un environnement multithread, vous devez d’abord créer un nouvel objet Session, puis créer l’objet Client ou Ressource à partir de Session. Pour plus d’informations, reportez-vous au Guide sur Sessions.

Lorsque vous appelez la boto3.resource(), vous utilisez implicitement l’objet Session par défaut. C’est pratique pour écrire du code à thread unique. Lorsque vous écrivez du code multithread, vous devez d’abord créer un nouvel objet Session pour chaque thread, puis récupérer la ressource de cet objet Session :

# Explicitly create a new Session for this 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 les délais d’expiration à une seconde :

import boto3
from botocore.config import Config

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

Pour d’autres discussions sur les délais d’expiration, consultez Tuning AWS Java SDK HTTP request settings for latency-aware DynamoDB applications. 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 version ultérieure, vous pouvez également contrôler TCP Keep-Alive :

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

import botocore
import boto3
from botocore.config import Config
from distutils.version import LooseVersion

required_version = "1.27.84"
current_version = botocore.__version__

my_config = Config(
   connect_timeout = 0.5,
   read_timeout = 0.5
)
if LooseVersion(current_version) > LooseVersion(required_version):
    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 adopter une vitesse de nouvelle tentative plus rapide 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 :

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 groupe différent est conservé par session. Si vous prévoyez que plus de 10 threads vont à l’encontre de clients ou de ressources créés à partir de la même session, vous devriez envisager de l’augmenter, afin que les threads n’aient pas à attendre d’autres threads utilisant une connexion groupée.

import boto3
from botocore.config import Config

my_config = Config(
   max_pool_connections = 20
)

# Setup a single session holding up to 20 pooled connections
session = boto3.Session(my_config)

# Create up to 20 resources against that session for handing to threads
# Notice the single-threaded access to the Session and each Resource
resource1 = session.resource('dynamodb')
resource2 = session.resource('dynamodb')
# etc

Gestion des erreurs

Les exceptions de service AWS ne sont pas toutes définies de manière statique dans Boto3. Cela est dû au fait que les erreurs et les exceptions liées aux services AWS varient considérablement et sont sujettes à changement. 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 :

Journalisation

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. L’argument 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 voyant 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

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 on Paginators et l’API reference for 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.