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

Pemrograman Amazon DynamoDB dengan Python dan Boto3

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

Tentang Boto

Anda dapat mengakses DynamoDB dari Python dengan menggunakan AWS SDK 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. Pustaka 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: Boto core dan Boto3.

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

Menggunakan dokumentasi Boto

Mulai 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 jenisnya ('S' untuk string, 'N' untuk angka) 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 ...
    }
)

Inilah PutItem operasi yang sama menggunakan antarmuka sumber daya. Pengetikan data implisit:

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 TypeDeserializer kelas TypeSerializer dan yang disediakan dengan 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()
    }

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 tabel 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 penulisan batch 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 kelompok. 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 membuat pengelola konteks untuk menulis objek dalam batch. Ini menyajikan antarmuka di mana sepertinya Anda menulis item satu per satu, tetapi secara internal itu menyangga dan mengirim item dalam batch. Ini juga menangani percobaan ulang item yang tidak 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

Jika Anda tidak membuat Sesi Anda sendiri, boto3.resource() dan boto3.client() menggunakan Sesi default bersama. Itu baik-baik saja dalam kode utas tunggal, tetapi dalam kode multi-utas setiap utas membangun objek dari Sesi yang sama, yang merupakan pola tidak aman yang dijelaskan sebelumnya.

Bagikan satu Klien di seluruh thread

Karena Klien aman untuk thread, pendekatan paling sederhana dan paling efisien adalah membuat satu Klien di utas utama Anda dan menyerahkannya kepada setiap pekerja. Mereka semua berbagi satu kumpulan koneksi.

Dibuat sekali, di utas utama, lalu dibagikan dengan semua utas:

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

Berikan setiap thread sumber dayanya sendiri

Karena Resource tidak aman untuk thread, setiap thread membutuhkannya sendiri. Buat Sesi baru di dalam thread dan buat Sumber Daya darinya, jadi tidak ada dua thread yang membangun Sumber Daya dari Sesi yang sama sekaligus.

Berjalan di dalam setiap utas:

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

Menyesuaikan objek Config

Saat membangun objek Client atau Resource, Anda dapat meneruskan parameter bernama opsional untuk menyesuaikan perilaku. Parameter bernama config membuka berbagai fungsi. Ini adalah contoh botocore.client.Config dan dokumentasi referensi untuk Config menunjukkan semua yang diekspos untuk Anda kontrol. Panduan untuk Konfigurasi memberikan gambaran yang baik.

Konfigurasi 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 bagi klien sebelum dapat mencoba lagi. Kode berikut mempersingkat batas waktu menjadi 10 detik:

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)

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

Konfigurasi untuk keep-alive

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

Menyetel 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

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)

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.

Konfigurasi untuk percobaan ulang

Konfigurasi juga menerima kamus yang disebut percobaan ulang di mana Anda dapat menentukan perilaku coba ulang yang Anda inginkan. Percobaan ulang terjadi di dalam SDK ketika SDK menerima kesalahan dan kesalahannya 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-nilai yang dapat Anda tentukan:

Definisi yang diperluas dari mode coba ulang ini dapat ditemukan di panduan untuk mencoba ulang serta dalam topik Perilaku C oba ulang di referensi SDK.

Berikut contoh yang secara eksplisit menggunakan kebijakan legacy coba ulang dengan maksimal 3 permintaan total (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 sesuai dengan kode Anda sendiri alih-alih mengandalkan boto3 untuk melakukan percobaan ulang implisit.

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

Untuk informasi tentang perubahan mendatang pada perilaku coba ulang default di semua AWS SDK, lihat Mengumumkan perilaku coba ulang yang diperbarui untuk AWS SDK dan Alat di Blog Alat Pengembang. AWS Untuk informasi selengkapnya tentang perubahan khususPython, lihat diskusi pembaruan perilaku coba lagi (#4789) di situs GitHub web.

Konfigurasi untuk koneksi pool maksimal

Terakhir, konfigurasi memungkinkan Anda mengontrol ukuran kumpulan koneksi:

Opsi ini mengontrol jumlah maksimum koneksi HTTP agar tetap dikumpulkan untuk digunakan kembali. Kumpulan yang berbeda disimpan untuk setiap Klien. Jika thread pada Klien menginginkan koneksi dan kumpulan kosong, itu akan membuat koneksi baru. Jika thread mencoba mengembalikan koneksi ke kolam setelah digunakan dan menemukan kolam penuh, itu akan menutup koneksi. Bila Anda mengantisipasi lebih dari 10 thread melawan satu Klien, pertimbangkan untuk menaikkan ukuran kumpulan agar sesuai dengan jumlah 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)

Setiap Resource juga memiliki kumpulan koneksi di dalamnya, tetapi objek Resource tidak aman untuk thread, jadi jangan bagikan di seluruh thread. Berikan setiap thread sumber dayanya sendiri.

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 ClientError a dan memperlihatkan 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 apa pun 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 pada klien Anda dan Anda menangkap pengecualian ini menggunakan instance klien Anda, seperti ini:

except ddb_client.exceptions.ProvisionedThroughputExceededException:

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

except ddb_resource.meta.client.exceptions.ProvisionedThroughputExceededException:

Untuk meninjau daftar jenis pengecualian yang 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. Pencatatan pesan yang kurang parah dari level akan diabaikan. Level logging termasukDEBUG,INFO,WARNING,ERROR, danCRITICAL. Nilai default-nya WARNING.

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

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

Cuplikan kode berikut menyetel sebagian besar logging ke INFO dengan DEBUG logging untuk titik akhir dan aktivitas 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)

Pengait acara

Botocore memancarkan peristiwa selama berbagai bagian pelaksanaannya. Anda dapat mendaftarkan penangan untuk peristiwa ini sehingga setiap kali 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 tabel DynamoDB apa pun di 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 parameter secara terprogram untuk mengubah perilaku:

params['TableName'] = "NewTableName"

Untuk informasi selengkapnya tentang acara, lihat dokumentasi botocore tentang peristiwa dan dokumentasi boto3 tentang peristiwa.

Pagination dan Paginator

Beberapa permintaan, seperti Query dan Scan, membatasi ukuran data yang dikembalikan pada satu permintaan dan mengharuskan Anda 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 ingin 10 item terakhir, Anda dapat menggunakan limit untuk mengambil hanya 10 yang terakhir. Perhatikan batasnya adalah berapa banyak yang harus dibaca dari tabel sebelum penyaringan diterapkan. Tidak ada cara untuk menentukan Anda ingin tepat 10 setelah pemfilteran; Anda hanya dapat mengontrol jumlah pra-filter 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, itu menunjukkan respons berakhir karena mencapai batas jumlah 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 itu, berarti tidak ada lagi item yang cocok dengan Query atau Scan.

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

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

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 Pelayan dan Referensi Pelayan.