Pemrograman Amazon DynamoDB dengan Python dan Boto3

Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.

Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.

Panduan ini memberikan orientasi kepada programmer yang ingin menggunakan Amazon DynamoDB dengan Python. Pelajari tentang berbagai lapisan abstraksi, manajemen konfigurasi, penanganan kesalahan, pengendalian kebijakan coba lagi, pengelolaan keep-alive, dan banyak lagi.

Tentang Boto

Anda dapat mengakses DynamoDB dari Python dengan menggunakan SDK AWS resmi untuk Python, yang biasa disebut sebagai Boto3. Nama Boto (diucapkan boh-toh) berasal dari lumba-lumba air tawar asli Sungai Amazon. Perpustakaan Boto3 adalah versi utama ketiga perpustakaan, pertama kali dirilis pada tahun 2015. Perpustakaan Boto3 cukup besar, karena mendukung semua AWS layanan, bukan hanya DynamoDB. Orientasi ini hanya menargetkan bagian Boto3 yang relevan dengan DynamoDB.

Boto dikelola dan diterbitkan oleh AWS sebagai proyek sumber terbuka yang dihosting di. GitHub Ini dibagi menjadi dua paket: Botocore dan Boto3.

Karena proyek ini di-host GitHub, Anda dapat melihat kode sumber, melacak masalah terbuka, atau mengirimkan masalah Anda sendiri.

Menggunakan dokumentasi Boto

Mulailah dengan dokumentasi Boto dengan sumber daya berikut:

Memahami lapisan abstraksi klien dan sumber daya

Dua antarmuka yang akan Anda gunakan adalah antarmuka klien dan antarmuka sumber daya.

Berikut adalah contoh menyisipkan item menggunakan antarmuka klien. Perhatikan bagaimana semua nilai dilewatkan sebagai peta dengan kunci yang menunjukkan tipenya ('S' untuk string, 'N' untuk nomor) dan nilainya sebagai string. Ini dikenal sebagai 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 ...
    }
)

Berikut adalah PutItem operasi yang sama menggunakan antarmuka sumber daya. Pengetikan data tersirat:

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

Jika diperlukan, Anda dapat mengonversi antara JSON biasa dan DynamoDB JSON menggunakan TypeSerializer kelas dan yang disediakan dengan 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()
    }

Berikut adalah cara melakukan query menggunakan antarmuka klien. Ini mengekspresikan query sebagai konstruksi JSON. Ini menggunakan KeyConditionExpression string yang membutuhkan substitusi variabel untuk menangani konflik kata kunci potensial:

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

Operasi kueri yang sama menggunakan antarmuka sumber daya dapat dipersingkat dan disederhanakan:

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

Sebagai contoh terakhir, bayangkan Anda ingin mendapatkan perkiraan ukuran tabel (yang merupakan metadata yang disimpan di atas meja yang diperbarui setiap 6 jam). Dengan antarmuka klien, Anda melakukan describe_table() operasi dan menarik jawaban dari struktur JSON yang dikembalikan:

import boto3

dynamodb = boto3.client('dynamodb')

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

Dengan antarmuka sumber daya, tabel melakukan operasi deskripsi secara implisit dan menyajikan data secara langsung sebagai atribut:

import boto3

dynamodb = boto3.resource('dynamodb')

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

Menggunakan sumber daya tabel batch_writer

Satu kenyamanan yang hanya tersedia dengan sumber daya tabel tingkat yang lebih tinggi adalah. batch_writer DynamoDB mendukung operasi batch write yang memungkinkan hingga 25 operasi put atau delete dalam satu permintaan jaringan. Batching seperti ini meningkatkan efisiensi dengan meminimalkan perjalanan pulang-pergi jaringan.

Dengan pustaka klien tingkat rendah, Anda menggunakan client.batch_write_item() operasi untuk menjalankan batch. Anda harus membagi pekerjaan Anda secara manual menjadi 25 batch. Setelah setiap operasi, Anda juga harus meminta untuk menerima daftar item yang belum diproses (beberapa operasi penulisan mungkin berhasil sementara yang lain bisa gagal). Anda kemudian harus meneruskan item yang belum diproses itu lagi ke batch_write_item() operasi selanjutnya. Ada sejumlah besar kode boilerplate.

Metode Table.batch_Writer menciptakan manajer konteks untuk menulis objek dalam batch. Ini menyajikan antarmuka di mana tampaknya seolah-olah Anda sedang menulis item satu per satu, tetapi secara internal itu buffering dan mengirim item dalam batch. Ini juga menangani percobaan ulang item yang belum diproses secara implisit.

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)

Contoh kode tambahan yang mengeksplorasi lapisan klien dan sumber daya

Anda juga dapat merujuk ke repositori contoh kode berikut yang mengeksplorasi penggunaan berbagai fungsi, menggunakan klien dan sumber daya:

Memahami bagaimana objek Klien dan Sumber Daya berinteraksi dengan sesi dan utas

Objek Resource tidak aman untuk utas dan tidak boleh dibagikan di seluruh utas atau proses. Lihat panduan tentang Sumber Daya untuk lebih jelasnya.

Objek Klien, sebaliknya, umumnya aman untuk utas, kecuali untuk fitur lanjutan tertentu. Lihat panduan tentang Klien untuk lebih jelasnya.

Objek Session tidak aman untuk utas. Jadi, setiap kali Anda membuat Klien atau Sumber Daya di lingkungan multi-utas, Anda harus membuat Sesi baru terlebih dahulu dan kemudian membuat Klien atau Sumber Daya dari Sesi. Lihat panduan tentang Sesi untuk lebih jelasnya.

Saat Anda memanggilboto3.resource(), Anda secara implisit menggunakan Sesi default. Ini nyaman untuk menulis kode single-threaded. Saat menulis kode multi-utas, Anda harus terlebih dahulu membuat Sesi baru untuk setiap utas dan kemudian mengambil sumber daya dari Sesi itu:

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

Menyesuaikan objek Config

Saat membuat objek Klien atau Sumber Daya, Anda dapat meneruskan parameter bernama opsional untuk menyesuaikan perilaku. Parameter bernama config membuka berbagai fungsi. Ini adalah instance dari botocore.client.Config dan dokumentasi referensi untuk Config menunjukkan semua yang diekspos untuk Anda kendalikan. Panduan untuk Konfigurasi memberikan gambaran yang baik.

Config untuk batas waktu

Salah satu penggunaan konfigurasi khusus adalah untuk menyesuaikan perilaku jaringan:

Batas waktu 60 detik berlebihan untuk DynamoDB. Ini berarti kesalahan jaringan sementara akan menyebabkan penundaan satu menit untuk klien sebelum dapat mencoba lagi. Kode berikut mempersingkat batas waktu menjadi satu detik:

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)

Untuk diskusi selengkapnya tentang batas waktu, lihat Menyetel setelan permintaan HTTP SDK AWS Java untuk aplikasi DynamoDB yang sadar latensi. Perhatikan bahwa Java SDK memiliki konfigurasi batas waktu lebih banyak daripada Python.

Config untuk keep-alive

Jika Anda menggunakan botocore 1.27.84 atau yang lebih baru, Anda juga dapat mengontrol TCP Keep-Alive:

Mengatur TCP Keep-Alive untuk True dapat mengurangi latensi rata-rata. Berikut contoh kode yang secara kondisional menetapkan TCP Keep-Alive ke true ketika Anda memiliki versi botocore yang tepat:

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)

Ada batasan berapa lama koneksi idle dapat tetap hidup. Pertimbangkan untuk mengirim permintaan berkala (katakanlah setiap menit) jika Anda memiliki koneksi idle tetapi ingin permintaan berikutnya menggunakan koneksi yang sudah dibuat.

Config untuk percobaan ulang

Konfigurasi juga menerima kamus yang disebut percobaan ulang di mana Anda dapat menentukan perilaku coba lagi yang Anda inginkan. Percobaan ulang terjadi dalam SDK ketika SDK menerima kesalahan dan kesalahan adalah tipe sementara. Jika kesalahan dicoba ulang secara internal (dan percobaan ulang akhirnya menghasilkan respons yang berhasil), tidak ada kesalahan yang terlihat dari perspektif kode panggilan, hanya latensi yang sedikit meningkat. Berikut adalah nilai yang dapat Anda tentukan:

Definisi yang diperluas dari mode coba lagi ini dapat ditemukan di panduan untuk mencoba ulang serta dalam topik perilaku Coba lagi dalam referensi SDK.

Berikut adalah contoh yang secara eksplisit menggunakan kebijakan legacy coba lagi dengan maksimal 3 total permintaan (2 percobaan ulang):

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)

Karena DynamoDB adalah sistem latensi rendah yang sangat tersedia, Anda mungkin ingin lebih agresif dengan kecepatan percobaan ulang daripada yang diizinkan oleh kebijakan coba ulang bawaan. Anda dapat menerapkan kebijakan coba ulang Anda sendiri dengan menyetel upaya maksimal ke 0, menangkap pengecualian sendiri, dan mencoba lagi sebagaimana mestinya dari kode Anda sendiri alih-alih mengandalkan boto3 untuk melakukan percobaan ulang implisit.

Jika Anda mengelola kebijakan coba ulang Anda sendiri, Anda akan ingin membedakan antara throttle dan error:

Config untuk koneksi kolam maksimal

Terakhir, konfigurasi memungkinkan Anda mengontrol ukuran kumpulan koneksi:

Opsi ini mengontrol jumlah maksimum koneksi HTTP agar tetap dikumpulkan untuk digunakan kembali. Kolam yang berbeda disimpan per Sesi. Jika Anda mengantisipasi lebih dari 10 utas yang bertentangan dengan klien atau sumber daya yang dibangun dari Sesi yang sama, Anda harus mempertimbangkan untuk meningkatkan ini, jadi utas tidak harus menunggu utas lain menggunakan koneksi gabungan.

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

Penanganan kesalahan

AWS pengecualian layanan tidak semuanya didefinisikan secara statis di Boto3. Ini karena kesalahan dan pengecualian dari AWS layanan sangat bervariasi dan dapat berubah. Boto3 membungkus semua pengecualian layanan sebagai a ClientError dan mengekspos detail sebagai JSON terstruktur. Misalnya, respons kesalahan mungkin terstruktur seperti ini:

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

Kode berikut menangkap ClientError pengecualian dan melihat nilai string di Code dalam Error untuk menentukan tindakan apa yang harus diambil:

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

Beberapa (tetapi tidak semua) kode pengecualian telah diwujudkan sebagai kelas tingkat atas. Anda dapat memilih untuk menangani ini secara langsung. Saat menggunakan antarmuka Klien, pengecualian ini diisi secara dinamis di klien Anda dan Anda menangkap pengecualian ini menggunakan instance klien Anda, seperti ini:

except ddb_client.exceptions.ProvisionedThroughputExceededException:

Saat menggunakan antarmuka Resource, Anda harus menggunakan .meta.client untuk melintasi dari sumber daya ke Klien yang mendasarinya untuk mengakses pengecualian, seperti ini:

except ddb_resource.meta.client.exceptions.ProvisionedThroughputExceededException:

Untuk meninjau daftar jenis pengecualian terwujud, Anda dapat membuat daftar secara dinamis:

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

Saat melakukan operasi tulis dengan ekspresi kondisi, Anda dapat meminta bahwa jika ekspresi gagal, nilai item harus dikembalikan dalam respons kesalahan.

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

Untuk bacaan lebih lanjut tentang penanganan kesalahan dan pengecualian:

Pencatatan log

Pustaka boto3 terintegrasi dengan modul logging bawaan Python untuk melacak apa yang terjadi selama sesi. Untuk mengontrol level logging, Anda dapat mengonfigurasi modul logging:

import logging

logging.basicConfig(level=logging.INFO)

Ini mengonfigurasi root logger untuk mencatat INFO dan pesan tingkat di atas. Pesan logging yang kurang parah dari level akan diabaikan. Tingkat logging termasukDEBUG,INFO,WARNING,ERROR, danCRITICAL. Nilai default-nya WARNING.

Logger di boto3 bersifat hierarkis. Pustaka menggunakan beberapa logger yang berbeda, masing-masing sesuai dengan bagian perpustakaan yang berbeda. Anda dapat mengontrol perilaku masing-masing secara terpisah:

Perpustakaan lain juga mencatat. Secara internal, boto3 menggunakan urllib3 pihak ketiga untuk penanganan koneksi HTTP. Ketika latensi penting, Anda dapat menonton lognya untuk memastikan kolam Anda dimanfaatkan dengan baik dengan melihat kapan urllib3 membuat koneksi baru atau menutup yang tidak aktif.

Cuplikan kode berikut menyetel sebagian besar logging INFO dengan DEBUG logging untuk aktivitas titik akhir dan kumpulan koneksi:

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)

Kait acara

Botocore memancarkan peristiwa selama berbagai bagian pelaksanaannya. Anda dapat mendaftarkan penangan untuk acara ini sehingga setiap kali suatu peristiwa dipancarkan, handler Anda akan dipanggil. Ini memungkinkan Anda memperluas perilaku botocore tanpa harus memodifikasi internal.

Misalnya, katakanlah Anda ingin melacak setiap kali PutItem operasi dipanggil pada setiap tabel DynamoDB dalam aplikasi Anda. Anda dapat mendaftar pada 'provide-client-params.dynamodb.PutItem' acara untuk menangkap dan mencatat setiap kali PutItem operasi dipanggil pada Sesi terkait. Inilah contohnya:

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

Di dalam handler, Anda bahkan dapat memanipulasi params secara terprogram untuk mengubah perilaku:

params['TableName'] = "NewTableName"

Untuk informasi lebih lanjut tentang acara, lihat dokumentasi botocore tentang acara dan dokumentasi boto3 tentang acara.

Pagination dan Paginator

Beberapa permintaan, seperti Query dan Scan, membatasi ukuran data yang dikembalikan pada satu permintaan dan mengharuskan Anda untuk membuat permintaan berulang untuk menarik halaman berikutnya.

Anda dapat mengontrol jumlah maksimum item yang akan dibaca untuk setiap halaman dengan limit parameter. Misalnya, jika Anda menginginkan 10 item terakhir, Anda dapat menggunakan limit untuk mengambil hanya 10 item terakhir. Perhatikan batasnya adalah berapa banyak yang harus dibaca dari tabel sebelum penyaringan apa pun diterapkan. Tidak ada cara untuk menentukan yang Anda inginkan tepat 10 setelah pemfilteran; Anda hanya dapat mengontrol jumlah yang telah difilter sebelumnya dan memeriksa sisi klien ketika Anda benar-benar mengambil 10. Terlepas dari batasnya, setiap respons selalu memiliki ukuran maksimum 1 MB.

Jika respons menyertakan aLastEvaluatedKey, ini menunjukkan respons berakhir karena mencapai batas hitungan atau ukuran. Kuncinya adalah kunci terakhir yang dievaluasi untuk respons. Anda dapat mengambil ini LastEvaluatedKey dan meneruskannya ke panggilan tindak lanjut ExclusiveStartKey untuk membaca potongan berikutnya dari titik awal itu. Ketika tidak ada yang LastEvaluatedKey dikembalikan, berarti tidak ada lagi item yang cocok dengan Kueri atau Pindai.

Berikut adalah contoh sederhana (menggunakan antarmuka Sumber Daya, tetapi antarmuka Klien memiliki pola yang sama) yang membaca paling banyak 100 item per halaman dan loop hingga semua item telah dibaca.

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

Untuk kenyamanan, boto3 dapat melakukan ini untuk Anda dengan Paginator. Namun, ini hanya berfungsi dengan antarmuka Klien. Berikut kode yang ditulis ulang untuk menggunakan Paginator:

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)

Untuk informasi selengkapnya, lihat Panduan tentang Paginator dan referensi API untuk DynamoDb.paginator.Query.

Pelayan

Pelayan memberikan kemampuan untuk menunggu sesuatu selesai sebelum melanjutkan. Saat ini, mereka hanya mendukung menunggu tabel dibuat atau dihapus. Di latar belakang, operasi pelayan melakukan pemeriksaan untuk Anda setiap 20 detik hingga 25 kali. Anda bisa melakukannya sendiri, tetapi menggunakan pelayan itu elegan saat menulis otomatisasi.

Kode ini menunjukkan cara menunggu tabel tertentu dibuat:

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

Untuk informasi lebih lanjut, lihat Panduan untuk Pelayan dan Referensi tentang Pelayan.