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. Scopri i diversi livelli di astrazione, la gestione della configurazione, la gestione degli errori, il controllo delle politiche di ripetizione dei tentativi, la gestione del keep-alive e altro ancora.

Informazioni su Boto

Puoi accedere a DynamoDB da Python utilizzando l'SDK AWS ufficiale per Python, comunemente noto come Boto3. Il nome Boto (pronunciato boh-toh) 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 AWS servizi, non solo DynamoDB. Questo orientamento 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 di Boto

Inizia con la documentazione di Boto con le seguenti risorse:

Comprensione dei livelli di astrazione del client e delle risorse

Le due interfacce con cui lavorerai sono l'interfaccia client e l'interfaccia delle risorse.

Ecco un esempio di inserimento di un elemento utilizzando l'interfaccia client. Notate come 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 è noto come formato 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 ...
    }
)

Ecco la stessa PutItem operazione utilizzando l'interfaccia delle risorse. La digitazione 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, puoi eseguire la conversione tra JSON normale e JSON di DynamoDB utilizzando TypeSerializer le classi e fornite con boto3: TypeDeserializer

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 client. Esprime la query come un costrutto JSON. Utilizza una KeyConditionExpression stringa che richiede la sostituzione di variabili per gestire eventuali conflitti di 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 interrogazione 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, immaginate 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 client, si esegue un'describe_table()operazione 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 tabellare di livello superiore è la. 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 batch migliora l'efficienza riducendo al minimo i round trip di rete.

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

Il metodo Table.batch_writer crea un gestore di contesto per scrivere oggetti in un batch. Presenta un'interfaccia in cui sembra che tu stia scrivendo gli elementi uno alla volta, ma internamente li memorizza nel buffer e li invia in batch. Inoltre, gestisce implicitamente i tentativi di nuovi articoli 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)

Esempi di codice aggiuntivi che esplorano i livelli client e resource

Puoi anche fare riferimento ai seguenti archivi di esempi di codice che esplorano l'utilizzo delle varie funzioni, utilizzando sia client che risorse:

Comprendere come gli oggetti Client e Resource interagiscono 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, ad eccezione di specifiche funzionalità avanzate. Per maggiori dettagli, consulta la guida sui client.

L'oggetto Session non è thread-safe. Quindi, ogni volta che crei un client o una risorsa in un ambiente multithread, dovresti prima creare una nuova sessione e poi creare il client o la risorsa dalla sessione. Consulta la guida sulle sessioni per maggiori dettagli.

Quando chiami ilboto3.resource(), stai usando implicitamente la Session predefinita. Questo è utile per scrivere codice a thread singolo. Quando scrivi codice multithread, vorrai prima costruire una nuova sessione per ogni thread e poi recuperare la risorsa da quella sessione:

# 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é tu possa controllarlo. La guida alla configurazione offre una buona panoramica.

Config 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. 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 ulteriori informazioni sui timeout, consulta Ottimizzazione delle impostazioni di richiesta HTTP di AWS Java SDK per le applicazioni DynamoDB che riconoscono la latenza. Nota che l'SDK Java ha più configurazioni di timeout rispetto a Python.

Config per keep-alive

Se stai usando botocore 1.27.84 o successivo, puoi anche controllare TCP Keep-Alive:

L'impostazione di TCP Keep-Alive su può ridurre le latenze medie. True Ecco un esempio di codice che imposta in modo condizionale TCP Keep-Alive su true quando hai la 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)

C'è un limite per quanto tempo 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.

Config per i nuovi tentativi

La configurazione accetta anche un dizionario chiamato retry in cui è possibile specificare il comportamento desiderato per i nuovi tentativi. I nuovi tentativi avvengono all'interno dell'SDK quando l'SDK riceve un errore e l'errore è di tipo transitorio. Se un errore viene ritentato internamente (e un nuovo tentativo alla fine produce una risposta corretta), non viene rilevato alcun errore dal punto di vista del codice di chiamata, ma solo una latenza leggermente elevata. Ecco i valori che puoi specificare:

Una definizione estesa di queste modalità di ripetizione è disponibile nella guida ai nuovi tentativi e nell'argomento sul comportamento Retry nel riferimento SDK.

Ecco un esempio che utilizza esplicitamente la politica di legacy riprova con un massimo di 3 richieste totali (2 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, potresti voler essere più aggressivo con la velocità dei nuovi tentativi rispetto a quanto consentito dalle politiche di ripetizione integrate. È possibile implementare la propria politica di nuovi tentativi impostando il numero massimo di tentativi su 0, rilevando personalmente le eccezioni e riprovando, se necessario, utilizzando il proprio codice invece di affidarsi a boto3 per eseguire nuovi tentativi impliciti.

Se gestisci la tua politica di nuovi tentativi, ti consigliamo di distinguere tra limitatori ed errori:

Config per il numero massimo di connessioni al 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 sessione. Se prevedi che più di 10 thread vadano contro client o risorse create sulla stessa sessione, dovresti prendere in considerazione la possibilità di generare 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

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 un file JSON strutturato ClientError ed espone i dettagli in formato 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 ClientError le eccezioni e analizza il valore della stringa all'Codeinterno 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. Puoi scegliere di gestirli direttamente. Quando si utilizza l'interfaccia Client, queste eccezioni vengono popolate dinamicamente sul client e le catture vengono rilevate utilizzando l'istanza client, in questo modo:

except ddb_client.exceptions.ProvisionedThroughputExceededException:

Quando si utilizza l'interfaccia Resource, è necessario passare dalla risorsa .meta.client 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, puoi generare l'elenco 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 fallisce, 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 integrato di Python per tracciare ciò che accade durante una sessione. Per controllare i livelli di registrazione, puoi configurare il modulo di registrazione:

import logging

logging.basicConfig(level=logging.INFO)

Questo configura il logger root per registrare INFO e registrare i messaggi di livello superiore. I messaggi di registrazione che sono meno severi del livello 2 verranno ignorati. I livelli di registrazione includonoDEBUG,, INFOWARNING, 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. Puoi controllare separatamente il comportamento di ciascuno:

Registrano anche altre librerie. 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 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 con la registrazione delle attività degli endpoint e dei pool di connessioni: INFO 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)

Ganci per 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 chiamato il gestore. Ciò consente di estendere il comportamento di botocore senza dover modificare i componenti interni.

Ad esempio, supponiamo di voler tenere traccia di ogni chiamata di un'PutItemoperazione su qualsiasi tabella DynamoDB dell'applicazione. È possibile registrarsi sull''provide-client-params.dynamodb.PutItem'evento per catturare e registrare ogni volta che viene richiamata un'PutItemoperazione sulla sessione 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, puoi persino manipolare i parametri a livello di codice per modificare il comportamento:

params['TableName'] = "NewTableName"

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

Pagination e Paginator

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

È possibile controllare il numero massimo di elementi da leggere per ogni pagina con il limit parametro. Ad esempio, se desideri gli ultimi 10 elementi, puoi utilizzare limit per recuperare solo gli ultimi 10. Nota che il limite è quanto deve essere letto dalla tabella prima di applicare qualsiasi filtro. Non c'è modo di specificare che desideri esattamente 10 dopo il filtraggio; puoi controllare il conteggio prefiltrato e controllare lato client solo quando ne hai effettivamente recuperati 10. Indipendentemente dal limite, ogni risposta ha sempre una dimensione massima di 1 MB.

Se la risposta include unLastEvaluatedKey, indica che la risposta è terminata perché ha raggiunto un limite di conteggio o dimensione. La chiave è l'ultima chiave valutata per la risposta. Puoi recuperarlo LastEvaluatedKey e passarlo a una chiamata successiva per leggere il blocco successivo da quel punto di partenza. ExclusiveStartKey Quando non viene LastEvaluatedKey restituito nulla, significa che non ci sono più elementi corrispondenti alla Query o alla 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ò farlo per te con Paginators. Tuttavia, funziona solo con l'interfaccia Client. Ecco il codice riscritto per usare Paginators:

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 all'API per DynamoDB.Paginator.Query.

Waiter

I camerieri offrono la possibilità di attendere che qualcosa venga completato prima di procedere. Al momento, supportano solo l'attesa della creazione o dell'eliminazione di un tavolo. In background, il cameriere effettua un controllo per voi ogni 20 secondi fino a 25 volte. Potresti farlo da solo, ma usare un cameriere è elegante quando si scrive l'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 camerieri e la Guida ai camerieri.