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 fournit une orientation aux programmeurs souhaitant 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 nouvelle tentative, la gestion du maintien en vie, etc.

À propos de Boto

Vous pouvez accéder à DynamoDB depuis Python en utilisant AWS le 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.

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

Commencez par consulter la documentation Boto avec les 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 des ressources.

Voici un exemple d'insertion d'un élément à l'aide de l'interface client. Remarquez comment 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 DynamoDB JSON.

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 PutItem opération en utilisant l'interface de ressources. Le typage 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 TypeDeserializer et fournies avec TypeSerializer 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. Il exprime la requête sous la forme d'une construction JSON. Il utilise une KeyConditionExpression chaîne 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 describe_table() opération 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 le. 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 cliente de bas niveau, vous utilisez l'client.batch_write_item()opération 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 batch_write_item() opération 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. Il présente une interface dans laquelle vous avez l'impression d'écrire des éléments un par un, mais en interne, il les met en mémoire tampon et les envoie par lots. Il gère également implicitement les nouvelles tentatives d'articles 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 de code supplémentaires qui explorent les couches client et ressource

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 :

Comprendre comment les objets client et ressource interagissent avec les sessions et les fils de discussion

L'objet Resource n'est pas adapté aux threads et ne doit pas être partagé entre les threads ou les processus. Reportez-vous au guide sur les ressources pour plus de détails.

L'objet client, en revanche, est généralement thread-safe, à l'exception de certaines fonctionnalités avancées. Reportez-vous au guide sur les clients pour plus de détails.

L'objet Session n'est pas adapté aux threads. Ainsi, chaque fois que vous créez un client ou une ressource dans un environnement multithread, vous devez d'abord créer une nouvelle session, puis créer le client ou la ressource à partir de la session. Reportez-vous au guide sur les sessions pour plus de détails.

Lorsque vous appelez leboto3.resource(), vous utilisez implicitement la 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 une nouvelle session pour chaque thread, puis récupérer la ressource de cette session :

# Explicitly create a new Session for this thread 
session = boto3.Session()
dynamodb = session.resource('dynamodb')

Personnalisation de l'objet Config

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 Config montre tout ce qu'il vous expose pour que vous puissiez le contrôler. Le guide de configuration fournit une bonne vue d'ensemble.

Config pour les délais

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

Les délais d'attente de 60 secondes sont excessifs pour DynamoDB. Cela signifie qu'un problème réseau transitoire retardera le client d'une minute avant qu'il ne puisse réessayer. Le code suivant réduit les délais d'attente à 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 en savoir plus sur les délais d'attente, 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 SDK Java possède plus de configurations de délai d'attente que Python.

Config 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 conditionnellement TCP Keep-Alive sur true lorsque vous avez 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)

La durée pendant laquelle une connexion inactive peut être maintenue en vie est limitée. 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.

Config pour les nouvelles tentatives

La configuration accepte également un dictionnaire appelé retries dans lequel vous pouvez spécifier le comportement de tentative souhaité. Les nouvelles tentatives se produisent dans le SDK lorsque celui-ci reçoit une erreur et que l'erreur est de type transitoire. Si une erreur est réessayée en interne (et qu'une nouvelle tentative produit finalement une réponse réussie), 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 la rubrique Comportement des tentatives dans la référence du SDK.

Voici un exemple qui utilise explicitement la politique de legacy nouvelle tentative avec un maximum de 3 demandes au total (2 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 relance plus rapide que ne le permettent les politiques de relance intégrées. Vous pouvez implémenter votre propre politique de 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 tentatives implicites.

Si vous gérez votre propre politique de nouvelles tentatives, vous devez faire la différence entre les limitations et les erreurs :

Config pour le nombre maximum de connexions au pool

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

Cette option contrôle le nombre maximum de connexions HTTP à conserver en pool en vue de leur réutilisation. Un pool 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

AWS les exceptions de service 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 AWS aux services varient considérablement et sont susceptibles d'être modifiées. Boto3 enveloppe toutes les exceptions de service sous forme de fichier JSON structuré ClientError et expose les détails sous forme de code 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 ClientError exceptions et examine la valeur de chaîne de caractères Code dans le 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 votre client et vous pouvez les intercepter à l'aide de votre instance cliente, comme suit :

except ddb_client.exceptions.ProvisionedThroughputExceededException:

Lorsque vous utilisez l'interface Resource, vous devez .meta.client 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'exceptions matérialisés, vous pouvez la générer dynamiquement :

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 conditionnelle, 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 :

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 enregistre INFO les messages de niveau supérieur ou supérieur. Les messages de journalisation moins sévères que le niveau seront ignorés. Les niveaux de journalisation incluent DEBUG INFOWARNING,ERROR,, etCRITICAL. L’argument par défaut est WARNING.

Les loggers 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 se connectent également. 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 de l'activité INFO du point de terminaison et du pool de connexions : DEBUG

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)

Crochets événementiels

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 PutItem opération est appelée sur une table DynamoDB de votre application. Vous pouvez vous enregistrer sur l''provide-client-params.dynamodb.PutItem'événement pour le catch et le journaliser chaque fois qu'une PutItem opération 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 la documentation de botocore sur les événements et la documentation de boto3 sur les événements.

Pagination et paginateur

Certaines demandes, telles que Query and Scan, limitent la taille des données renvoyées lors d'une seule demande et vous obligent à effectuer 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 limit paramètre. Par exemple, si vous voulez obtenir les 10 derniers éléments, vous pouvez les utiliser limit pour récupérer uniquement les 10 derniers. Notez que la limite est celle qui doit être lue dans le tableau avant qu'un filtrage ne soit appliqué. Il n'est pas possible de spécifier que vous voulez exactement 10 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 récupéré 10. Quelle que soit la limite, chaque réponse a toujours une taille maximale de 1 Mo.

Si la réponse inclut unLastEvaluatedKey, 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 le récupérer LastEvaluatedKey et le transmettre à un appel de suivi ExclusiveStartKey afin de lire le morceau suivant à partir de ce point de départ. Si aucun résultat n'est LastEvaluatedKey renvoyé, cela signifie qu'il n'y a plus d'éléments correspondant à la requête ou au scan.

Voici un exemple simple (utilisant l'interface Resource, mais l'interface client suit le même schéma) qui lit au maximum 100 éléments par page et fait une 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 Paginators. 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 sur les paginateurs et la référence d'API pour DynamoDB.Paginator.Query.

Programmes d'attente

Les serveurs 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 serveur 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 serveur est élégante lors de l'automatisation de l'écriture.

Ce code montre comment attendre qu'une table particulière soit 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 des serveurs et le document de référence sur les serveurs.