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

È possibile accedere a DynamoDB da Python utilizzando l’SDK AWS ufficiale per Python, comunemente noto come 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 AWS, non solo DynamoDB. Questa guida si rivolge solo alle parti di Boto3 rilevanti per DynamoDB.

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

Poiché questi progetti sono ospitati su GitHub, è possibile visualizzare il codice sorgente, tenere traccia dei problemi aperti o inviare i propri 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, è necessario anche richiedere di ricevere un elenco di elementi non elaborati (alcune operazioni di scrittura potrebbero avere successo mentre altre potrebbero non andare a buon fine). È 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 scrivere 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

L’oggetto Resource non è thread-safe e non deve essere condiviso tra thread o processi. Consulta la guida su Resource per maggiori dettagli.

L’oggetto Client, al contrario, è generalmente thread-safe, a eccezione di specifiche funzionalità avanzate. Consulta la guida su Client per maggiori dettagli.

L’oggetto Session non è thread-safe. Pertanto, ogni volta che si creano Client o Resource in un ambiente multi-thread, prima sarebbe opportuno creare una nuova Session e quindi creare Client o Resource da Session. Consulta la guida su Session per maggiori dettagli.

Quando si chiama boto3.resource(), si sta implicitamente usando la Session predefinita. Ciò è utile per scrivere codice a thread singolo. In caso di scrittura di codice multi-thread, sarà necessario prima costruire una nuova Session per ogni thread e poi recuperare la risorsa da quella Session:

# Explicitly create a new Session for this 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 un secondo:

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)

Per un’utile discussione su questo argomento, consulta Tuning AWS Java SDK HTTP request settings for latency-aware DynamoDB applications. L’SDK per Java ha più configurazioni di timeout rispetto a Python.

Configurazione per keep-alive

Se si sta usando botocore 1.27.84 o successivo, è possibile anche controllare TCP Keep-Alive:

L’impostazione di TCP Keep-Alive su True può ridurre le latenze medie. Ecco un esempio di codice che imposta in base a un condizione il Keep-Alive su true quando si dispone della versione botocore corretta:

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)

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 bassa latenza, potrebbe essere possibile volere maggiore aggressività con la velocità della ripetizione dei tentativi rispetto a quanto consentito dalle policy di ripetizione dei tentativi 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:

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 mantenuto un pool diverso per Session. Se si prevede che più di 10 thread utilizzino client o risorse creati sulla stessa Session, sarebbe opportuno prendere in considerazione la possibilità di aumentare questo valore, in modo che i thread non debbano attendere che altri thread utilizzino una connessione condivisa.

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

Gestione degli errori

Le eccezioni dei servizi AWS non sono tutte definite staticamente in Boto3. Ciò avviene perché gli errori e le eccezioni dei servizi AWS 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

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, è possibile controllarne i log per assicurarsi che il pool venga utilizzato bene, vedendo 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

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 sugli Impaginatori e la documentazione di riferimento dell’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.