Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.

Programmazione di Amazon DynamoDB con Python e Boto3

Questa guida fornisce un orientamento ai programmatori che desiderano utilizzare Amazon DynamoDB con Python. La guida copre i diversi livelli di astrazione, la gestione della configurazione, la gestione degli errori, il controllo delle policy di ripetizione dei tentativi, la gestione del keep-alive e altro ancora.

Informazioni su Boto

Puoi accedere a DynamoDB da Python utilizzando l' AWS SDK ufficiale per Python, comunemente chiamato Boto3. Il nome Boto deriva da un delfino d’acqua dolce originario del Rio delle Amazzoni. La libreria Boto3 è la terza versione principale della libreria, rilasciata per la prima volta nel 2015. La libreria Boto3 è piuttosto ampia, in quanto supporta tutti i servizi, non solo DynamoDB. AWS Questa guida si rivolge solo alle parti di Boto3 rilevanti per DynamoDB.

Boto è gestito e pubblicato da un progetto open source ospitato AWS su. GitHub È suddiviso in due pacchetti: Botocore e Boto3.

Poiché questi progetti sono ospitati su GitHub, puoi visualizzare il codice sorgente, tenere traccia dei problemi aperti o inviare i tuoi problemi.

Utilizzo della documentazione Boto

Inizia a esplorare la documentazione Boto con le seguenti risorse:

Informazioni sui livelli di astrazione del client e delle risorse

Le due interfacce di lavoro sono l’interfaccia del client e l’interfaccia delle risorse.

Ecco un esempio di inserimento di un elemento utilizzando l’interfaccia del client. Tutti i valori vengono passati come mappa con la chiave che ne indica il tipo (“S” per stringa, “N” per numero) e il loro valore come stringa. Questo approccio è noto come formato 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 ...
    }
)

Ecco la stessa operazione PutItem utilizzando l’interfaccia delle risorse. La tipizzazione dei dati è implicita:

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 ...
    }
)

Se necessario, è possibile eseguire la conversione tra JSON normale e JSON DynamoDB utilizzando le classi TypeSerializer e TypeDeserializer fornite con 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()
    }

Ecco come eseguire una query utilizzando l’interfaccia del client. Esprime la query come un costrutto JSON. Utilizza una stringa KeyConditionExpression che richiede la sostituzione di variabili per gestire eventuali conflitti tra parole chiave:

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 stessa operazione di query utilizzando l’interfaccia delle risorse può essere abbreviata e semplificata:

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')
)

Come ultimo esempio, si immagini di voler ottenere la dimensione approssimativa di una tabella (ossia i metadati conservati nella tabella che viene aggiornata ogni 6 ore circa). Con l’interfaccia del client, si esegue un’operazione describe_table() e si estrae la risposta dalla struttura JSON restituita:

import boto3

dynamodb = boto3.client('dynamodb')

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

Con l’interfaccia delle risorse, la tabella esegue l’operazione di descrizione in modo implicito e presenta i dati direttamente come attributo:

import boto3

dynamodb = boto3.resource('dynamodb')

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

Utilizzo della risorsa della tabella batch_writer

Una comodità disponibile solo con la risorsa della tabella di livello superiore è batch_writer. DynamoDB supporta operazioni di scrittura in batch che consentono fino a 25 operazioni di inserimento o eliminazione in una richiesta di rete. Questo tipo di operazioni in batch migliora l’efficienza riducendo al minimo i round trip di rete.

Con la libreria client di basso livello è possibile utilizzare l’operazione client.batch_write_item() per eseguire batch. È necessario suddividere manualmente il lavoro in batch da 25. Dopo ogni operazione, è inoltre necessario richiedere di ricevere un elenco di elementi non elaborati (alcune operazioni di scrittura potrebbero avere esito positivo mentre altre potrebbero fallire). È quindi necessario passare nuovamente gli elementi non elaborati a un’operazione batch_write_item() successivamente. È presente una quantità significativa di codice boilerplate.

Il metodo Table.batch _writer crea un gestore di contesto per la scrittura di oggetti in un batch. Questo presenta un’interfaccia in cui sembra che si stiano scrivendo gli elementi uno alla volta, ma internamente il gestore li memorizza nel buffer e li invia in batch. Gestisce inoltre implicitamente la ripetizione dei tentativi per gli elementi non elaborati.

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)

Ulteriori esempi di codice che esplorano i livelli del client e delle risorse

È possibile anche fare riferimento ai seguenti repository di esempi di codice che esplorano l’utilizzo delle varie funzioni, utilizzando sia client che risorse:

Informazioni sull’interazione degli oggetti Client e Resource con sessioni e thread

Se non crei la tua sessione boto3.resource() e boto3.client() usi una sessione predefinita condivisa. Questo va bene nel codice a thread singolo, ma nel codice a più thread ogni thread crea oggetti dalla stessa sessione, che è il modello non sicuro descritto in precedenza.

Condividi un client tra i thread

Poiché un cliente è thread-safe, l'approccio più semplice ed efficiente è creare un singolo client nel thread principale e consegnarlo a tutti i lavoratori. Condividono tutti il suo unico pool di connessioni.

Creato una volta, nel thread principale, quindi condiviso con tutti i thread:

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

Assegna a ogni thread la propria risorsa

Poiché una risorsa non è thread-safe, ogni thread ha bisogno della propria. Crea una nuova sessione all'interno del thread e crea la risorsa da essa, in modo che due thread non generino risorse dalla stessa sessione contemporaneamente.

Viene eseguito all'interno di ogni thread:

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

Personalizzazione dell’oggetto Config

Quando si costruisce un oggetto Client o Resource, è possibile passare parametri denominati opzionali per personalizzare il comportamento. Il parametro denominato config sblocca una serie di funzionalità. È un’istanza di botocore.client.Config e la documentazione di riferimento per Config mostra tutto ciò che espone affinché sia possibile controllarlo. La guida alla configurazione offre una buona panoramica.

Configurazione per i timeout

Un uso di una configurazione personalizzata è quello di regolare i comportamenti di rete:

I timeout di 60 secondi sono eccessivi per DynamoDB. Ciò significa che un problema temporaneo della rete causerà un minuto di ritardo per il client prima di poter riprovare. Il codice seguente riduce i timeout a 10 secondi:

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)

Per ulteriori informazioni sui timeout, consulta Ottimizzazione delle impostazioni di richiesta HTTP di AWS Java SDK per applicazioni DynamoDB sensibili alla latenza. L’SDK per Java ha più configurazioni di timeout rispetto a Python.

Configurazione per keep-alive

Se usi botocore 1.27.84 o versione successiva, puoi anche controllare il protocollo TCP: Keep-Alive

L'impostazione di TCP su può ridurre le latenze medie. Keep-Alive True Ecco un esempio di codice che imposta condizionalmente TCP Keep-Alive su true quando si dispone della versione botocore corretta:

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)

Esiste un limite per la quantità di tempo in cui una connessione inattiva può essere mantenuta attiva. Prendi in considerazione l’invio di richieste periodiche, ad esempio ogni minuto, se hai una connessione inattiva ma desideri che la richiesta successiva utilizzi una connessione già stabilita.

Configurazione per le ripetizioni di tentativi

La configurazione accetta anche un dizionario denominato Nuovi tentativi in cui è possibile specificare il comportamento desiderato per la ripetizione dei tentativi. Le ripetizioni di tentativi avvengono all’interno dell’SDK quando l’SDK riceve un errore e l’errore è di tipo transitorio. Se dopo un errore viene effettuata una ripetizione di tentativo internamente, successivamente seguita da una risposta positiva), dal punto di vista del codice non c’è alcun errore, ma solo una latenza leggermente maggiore. Ecco i valori che è possibile specificare:

Una definizione estesa di queste modalità di ripetizione dei tentativi è disponibile nella guida alla ripetizione dei tentativi e nell’argomento Retry behavior nella documentazione di riferimento dell’SDK.

Ecco un esempio che utilizza esplicitamente la policy di ripetizione dei tentativi legacy con un massimo di tre richieste totali (due ripetizioni di tentativi):

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)

Poiché DynamoDB è un sistema ad alta disponibilità e a bassa latenza, potresti voler essere più aggressivo con la velocità dei tentativi rispetto a quella consentita dalle politiche di ripetizione integrate. È possibile implementare la propria policy di ripetizione dei tentativi impostando il numero massimo di tentativi su 0, individuando personalmente le eccezioni ed effettuando ripetizioni dei tentativi, se necessario, utilizzando il proprio codice invece di affidarsi a boto3 per eseguire ripetizioni di tentativi implicite.

Se si gestisce in autonomia la propria policy sulla ripetizione dei tentativi, si consiglia di distinguere tra limitazioni (della larghezza di banda della rete) ed errori:

Per informazioni sulle modifiche imminenti al comportamento di ripetizione predefinito in tutti gli SDK, consulta Annuncio del comportamento di riprova aggiornato per AWS SDK and Tools nel Developer Tools Blog. AWS AWS Per ulteriori informazioni sulle modifiche specifiche diPython, consulta la discussione sull'aggiornamento del comportamento dei tentativi (#4789) sul sito Web. GitHub

Configurazione per il numero massimo di connessioni nel pool

Infine, la configurazione consente di controllare la dimensione del pool di connessioni:

Questa opzione controlla il numero massimo di connessioni HTTP da conservare in pool per il riutilizzo. Viene creato un pool diverso per ogni cliente. Se un thread su un Client desidera una connessione e il pool è vuoto, effettuerà una nuova connessione. Se un thread tenta di restituire una connessione al pool dopo l'uso e trova il pool pieno, chiuderà la connessione. Se prevedi che più di 10 thread siano indirizzati a un singolo client, valuta la possibilità di aumentare le dimensioni del pool in modo che corrispondano al numero di thread.

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)

Ogni risorsa ha anche un pool di connessioni all'interno, ma un oggetto Resource non è thread-safe, quindi non condividerne uno tra i thread. Assegna a ogni thread la propria risorsa.

Gestione degli errori

AWS le eccezioni di servizio non sono tutte definite staticamente in Boto3. Questo perché gli errori e le eccezioni AWS dei servizi variano notevolmente e sono soggetti a modifiche. Boto3 racchiude tutte le eccezioni di servizio come ClientError ed espone i dettagli come JSON strutturato. Ad esempio, una risposta di errore potrebbe essere strutturata in questo modo:

{
    '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
    }
}

Il codice seguente rileva tutte eccezioni ClientError e analizza il valore della stringa di Code all’interno di Error per determinare l’azione da intraprendere:

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

Alcuni (ma non tutti) i codici di eccezione sono stati materializzati come classi di primo livello. È possibile scegliere di gestirli direttamente. Quando si utilizza l’interfaccia Client, queste eccezioni vengono popolate dinamicamente sul client ed è possibile intercettarle utilizzando l’istanza client in questo modo:

except ddb_client.exceptions.ProvisionedThroughputExceededException:

Quando si utilizza l’interfaccia Resource, è necessario utilizzare .meta.client per passare dalla risorsa al Client sottostante per accedere alle eccezioni, in questo modo:

except ddb_resource.meta.client.exceptions.ProvisionedThroughputExceededException:

Per esaminare l’elenco dei tipi di eccezioni materializzate, è possibile generarlo dinamicamente:

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

Quando si esegue un’operazione di scrittura con un’espressione condizionale, è possibile richiedere che, se l’espressione non va a buon fine, il valore dell’elemento venga restituito nella risposta di errore.

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'])

Per ulteriori informazioni sulla gestione degli errori e sulle eccezioni:

Registrazione dei log

La libreria boto3 si integra con il modulo di registrazione di log integrato di Python per tracciare ciò che accade durante una sessione. Per controllare i livelli di registrazione di log, è possibile configurare il modulo relativo:

import logging

logging.basicConfig(level=logging.INFO)

Questo configura il logger root per registrare log di messaggi di livello INFO e superiore. I messaggi di registrazione di log che sono meno gravi di questo livello verranno ignorati. I livelli di logging includono DEBUG, INFO, WARNING, ERROR e CRITICAL. Il valore predefinito è WARNING.

I logger in boto3 sono gerarchici. La libreria utilizza alcuni logger diversi, ciascuno corrispondente a diverse parti della libreria. È possibile controllare separatamente il comportamento di ciascuno:

Anche altre librerie effettuano la registrazione di log. Internamente, boto3 utilizza l’urllib3 di terze parti per la gestione della connessione HTTP. Quando la latenza è importante, puoi controllarne i log per assicurarti che il pool sia ben utilizzato, osservando quando urllib3 stabilisce una nuova connessione o ne chiude una inattiva.

Il seguente frammento di codice imposta la maggior parte della registrazione di log su INFO con la registrazione di log di livello DEBUG per l’attività degli endpoint e del pool di connessioni:

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)

Hook degli eventi

Botocore emette eventi durante varie fasi della sua esecuzione. È possibile registrare i gestori per questi eventi in modo che ogni volta che viene emesso un evento venga richiamato il gestore. Ciò consente di estendere il comportamento di botocore senza dover modificare i componenti interni.

Ad esempio, si supponga di voler tenere traccia di ogni chiamata di un’operazione PutItem su qualsiasi tabella DynamoDB dell’applicazione. È possibile registrarsi all’evento 'provide-client-params.dynamodb.PutItem' per l’intercettazione e la registrazione di log ogni volta che viene invocata un’operazione PutItem sulla Session associata. Ecco un esempio:

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 ...
    }
)

All’interno del gestore, è possibile anche manipolare i parametri a livello di codice per modificare il comportamento:

params['TableName'] = "NewTableName"

Per ulteriori informazioni sugli eventi, consulta la documentazione botocore sugli eventi e la documentazione boto3 sugli eventi.

Impaginazione e impaginatore

Alcune richieste, come Query e Scan, limitano la dimensione dei dati restituiti in una singola richiesta e richiedono l’esecuzione di richieste ripetute per richiamare le pagine successive.

È possibile controllare il numero massimo di elementi da leggere per ogni pagina con il parametro limit. Ad esempio, se si desiderano gli ultimi 10 elementi, è possibile utilizzare limit per recuperare solo questi ultimi. Il limite è la quantità che deve essere letta dalla tabella prima di applicare qualsiasi filtro. Non c’è modo di specificare che si desiderano esattamente 10 elementi dopo il filtraggio; è possibile controllare il conteggio prefiltrato e controllare lato client solo quando sono stati effettivamente recuperati 10 elementi. Indipendentemente dal limite, ogni risposta ha sempre una dimensione massima di 1 MB.

Se la risposta include una LastEvaluatedKey, ciò indica che la risposta è terminata perché ha raggiunto un limite di conteggio o dimensione. La chiave è l’ultima chiave valutata per la risposta. È possibile recuperare questa LastEvaluatedKey e passarla a una chiamata successiva come ExclusiveStartKey in modo da leggere il blocco successivo da quel punto di partenza. Quando non viene restituita nessuna LastEvaluatedKey, significa che non ci sono più elementi corrispondenti a Query o Scan.

Ecco un semplice esempio (utilizzando l’interfaccia Resource, ma l’interfaccia Client ha lo stesso schema) che legge al massimo 100 elementi per pagina e si ripete fino a quando tutti gli elementi non sono stati letti.

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']

Per comodità, boto3 può occuparsi di questa operazione attraverso gli Impaginatori. Tuttavia, l’opzione funziona solo con l’interfaccia Client. Ecco il codice riscritto per usare gli Impaginatori:

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)

Per ulteriori informazioni, consulta la Guida sui paginatori e il riferimento API per. DynamoDB.Paginator.Query

Waiter

I waiter offrono la possibilità di attendere che un’operazione venga completata prima di procedere. Al momento, supportano solo l’attesa della creazione o dell’eliminazione di una tabella. In background, il waiter effettua un controllo ogni 20 secondi fino a 25 volte. È possibile fare questa operazione in autonomia, ma usare un waiter rende elegante il codice di automazione.

Questo codice mostra come attendere la creazione di una particolare tabella:

# 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'])

Per ulteriori informazioni, consulta la Guida ai waiter e il Riferimento sui waiter.