Python と Boto3 による Amazon DynamoDB のプログラミング

このガイドは、Python で Amazon DynamoDB を使用したいと考えているプログラマーを対象としています。さまざまな抽象化レイヤー、設定管理、エラー処理、再試行ポリシーの制御、キープアライブの管理などについて説明します。

Boto について

Python から DynamoDB にアクセスするには、Python 用の公式 AWS SDK (一般に Boto3 と呼ばれる) を使用します。Boto (ボトと発音) という名前は、アマゾン川に自生する淡水イルカに由来しています。Boto3 ライブラリは, ライブラリの 3 番目のメジャーバージョンであり、2015 年に初めてリリースされました。Boto3 ライブラリは、DynamoDB だけでなくすべての AWS サービスをサポートするため、非常に大きいです。このオリエンテーションは、DynamoDB に関連する Boto3 の部分のみを対象としています。

Boto は、GitHub でホストされるオープンソースプロジェクトとして AWS によって管理および公開されています。Botocore と Boto3 の 2 つのパッケージに分かれています。

これらのプロジェクトは GitHub でホストされているため、ソースコードを表示したり、未解決の問題を追跡したり、独自の問題を送信したりできます。

Boto ドキュメントの使用

以下のリソースを使って Boto ドキュメントを使い始めましょう。

クライアントとリソースの抽象化レイヤーを理解する

使用する 2 つのインターフェイスは、クライアントインターフェイスとリソースインターフェイスです。

クライアントインターフェイスを使用して項目を挿入する例を次に示します。すべての値が、型を示すキー (文字列は 'S'、数値は 'N') と文字列としての値のマップとして渡されることに注目してください。これは 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 ...
    }
)

リソースインターフェイスを使用した同じ PutItem 操作を次に示します。データ型の指定は暗黙的です。

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

必要に応じて、boto3 に用意されている TypeSerializer および TypeDeserializer クラスを使用して、通常の JSON と DynamoDB JSON の間で変換できます。

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

クライアントインターフェイスを使用してクエリを実行する方法は次のとおりです。クエリは JSON コンストラクトとして表現されます。キーワードの競合が発生する可能性がある場合は、変数置換が必要な KeyConditionExpression 文字列を使用します。

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

リソースインターフェイスを使った同じクエリ操作を短くして簡略化できます。

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

最後の例として、テーブルのおおよそのサイズ (テーブルに保持され、約 6 時間ごとに更新されるメタデータ) を取得したいとします。クライアントインターフェイスで、describe_table() 操作を行い、返された JSON 構造から回答を引き出します。

import boto3

dynamodb = boto3.client('dynamodb')

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

リソースインターフェイスでは、テーブルは describe 操作を暗黙的に実行し、データを属性として直接表示します。

import boto3

dynamodb = boto3.resource('dynamodb')

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

テーブルリソース batch_writer の使用

上位レベルのテーブルリソースでのみ利用できる便利な機能の 1 つは、batch_writer です。DynamoDB はバッチ書き込み操作をサポートしており、1 つのネットワークリクエストで最大 25 件の入力または削除操作が可能です。このようなバッチ処理は、ネットワークラウンドトリップを最小限に抑えることで効率性を高めます。

低レベルのクライアントライブラリでは、client.batch_write_item() 操作を使用してバッチを実行します。作業は手動で 25 個のバッチに分割する必要があります。各操作の後に、未処理の項目のリストを受け取るように要求する必要もあります (書き込み操作の中には成功するものもあれば、失敗するものもあります)。その後、それらの未処理項目を後の batch_write_item() 操作に再度渡す必要があります。大量のボイラープレートコードがあります。

Table.batch_writer メソッドは、オブジェクトをバッチで書き込むためのコンテキストマネージャーを作成します。一度に 1 つずつ項目を書き込んでいるように見えても、内部的には項目をバッファリングしてバッチで送信するインターフェイスを提供します。また、未処理の項目の再試行も暗黙的に処理されます。

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)

クライアントレイヤーとリソースレイヤーを調べるその他のコード例

次のコードサンプルリポジトリを参照して、クライアントとリソースの両方を使用してさまざまな機能の使用法を調べることもできます。

Client オブジェクトと Resource オブジェクトがセッションやスレッドとどのように相互作用するかを理解する

独自のセッションを作成しない場合、boto3.resource() と boto3.client() は共有デフォルト Session を使用します。これはシングルスレッドのコードでは問題ありませんが、マルチスレッドのコードでは、すべてのスレッドが同じ Session からオブジェクトを構築することになるため、先ほど説明した安全ではないパターンになります。

スレッド間で 1 つの Client を共有する

Client はスレッドセーフであるため、最もシンプルで効率的なアプローチは、メインスレッドで単一の Client を作成し、すべてのワーカーに渡すことです。これらはすべて 1 つの接続プールを共有します。

メインスレッドで 1 回作成され、すべてのスレッドと共有されます。

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

各スレッドに独自の Resource を付与する

Resource はスレッドセーフではないため、各スレッドには独自の Resource が必要です。スレッド内に新しい Session を作成し、そこから Resource を作成するため、2 つのスレッドが同じ Session から同時に Resource を構築することはありません。

各スレッド内で実行されます。

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

Config オブジェクトのカスタマイズ

Client または Resource オブジェクトを作成するときに、オプションの名前付きパラメータを渡して動作をカスタマイズできます。config という名前のパラメータは、さまざまな機能をアンロックします。これは botocore.client.Config のインスタンスであり、Config のリファレンスドキュメントには、ユーザーが制御できるすべての情報が記載されています。設定ガイドには、概要がわかりやすく書かれています。

タイムアウトの設定

カスタム設定の用途の 1 つは、ネットワークの動作を調整することです。

DynamoDB では 60 秒のタイムアウトは過剰です。つまり、一時的なネットワークの不具合により、クライアントが再試行できるようになるまでに 1 分間の遅延が生じます。次のコードはタイムアウトを 10 秒に短縮します。

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)

タイムアウトの詳細については、「レイテンシーを考慮した DynamoDB アプリケーションのための AWS Java SDK HTTP リクエスト設定のチューニング」を参照してください。Java SDK には Python よりも多くのタイムアウト設定があることに注意してください。

キープアライブの設定

botocore 1.27.84 以降を使用している場合は、TCP キープアライブを制御することもできます。

TCP キープアライブを True に設定すると、平均レイテンシーを減らすことができます。以下は、適切な botocore バージョンを使用している場合に TCP キープアライブを条件付きで true に設定するサンプルコードです。

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)

アイドル状態の接続を維持できる時間には制限があります。接続がアイドル状態で、次のリクエストではすでに確立されている接続を使いたい場合は、定期的に (たとえば 1 分ごとに) リクエストを送信することを検討してください。

再試行の設定

この設定では、希望する再試行動作を指定できる retries という辞書も使用できます。SDK がエラーを受信し、そのエラーが一時的なタイプの場合、SDK 内で再試行が行われます。エラーが内部で再試行された場合 (そして再試行によって最終的に成功の応答が得られた場合)、呼び出し側のコードから見ると、エラーは発生せず、レイテンシーがわずかに増加するだけです。指定できる値は次のとおりです。

これらの再試行モードの拡張定義は、再試行ガイドと SDK リファレンスの「再試行動作」トピックに記載されています。

次の例は、legacy 再試行ポリシーを明示的に使用して、合計リクエスト 3 回 (再試行 2 回) までにした例です。

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)

DynamoDB は可用性が高く、レイテンシーが低いシステムであるため、組み込みの再試行ポリシーで許可されているよりも積極的に再試行の速度を上げる必要がある場合があります。boto3 に頼って暗黙的な再試行を行うのではなく、最大試行回数を 0 に設定し、自分で例外をキャッチし、必要に応じて独自のコードから再試行することで、独自の再試行ポリシーを実装できます。

独自の再試行ポリシーを管理する場合は、スロットルとエラーを区別しておきましょう。

すべての AWS SDK のデフォルトの再試行動作に対する今後の変更については、AWS デベロッパーツールブログの「AWS SDK およびツールの更新された再試行動作の発表」を参照してください。Python に固有の変更の詳細については、GitHub ウェブサイトの「再試行動作更新の説明 (#4789)」を参照してください。

最大プール接続数の設定

最後に、この構成では接続プールのサイズを制御できます。

このオプションは、プールして再利用できるように保持する HTTP 接続の最大数を制御します。Client ごとに異なるプールが保持されます。Client のスレッドが接続を必要としており、プールが空の場合、新しい接続が作成されます。スレッドが使用後にプールへ接続を返そうとした際、プールがいっぱいであることがわかった場合、接続は閉じられます。1 つの Client に対して 10 を超えるスレッドが発生することが予想される場合は、スレッド数に合わせてプールサイズを増やすことを検討してください。

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)

各 Resource にも接続プールがありますが、Resource オブジェクトはスレッドセーフではないため、スレッド間で共有しないでください。各スレッドに独自の Resource を付与します。

エラー処理

AWSBoto3 では サービスの例外がすべて静的に定義されているわけではありません。これは、AWS サービスのエラーや例外は大きく異なり、変更される可能性があるためです。Boto3 はすべてのサービス例外を ClientError としてラップし、詳細を構造化された JSON として公開します。例えば、エラーレスポンスは以下のような構造になっているかもしれません。

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

次のコードは、すべての ClientError 例外をキャッチし、Error 内の Code の文字列値を調べて実行するアクションを決定します。

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

一部の (すべてではない) 例外コードは最上位クラスとして実装されています。これらを直接処理することもできます。Client インターフェイスを使用する場合、これらの例外はクライアントに動的に入力され、以下のようにクライアントインスタンスを使用してこれらの例外をキャッチします。

except ddb_client.exceptions.ProvisionedThroughputExceededException:

Resource インターフェイスを使用しているときには、以下のように、.meta.client を使用してリソースから基盤となる Client までトラバースして例外にアクセスする必要があります。

except ddb_resource.meta.client.exceptions.ProvisionedThroughputExceededException:

マテリアライズド例外タイプのリストを確認するには、リストを動的に生成できます。

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

条件式を使用して書き込み操作を行う場合、式が失敗した場合にその項目の値をエラーレスポンスで返すようにリクエストできます。

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

エラー処理と例外について詳しくは、以下を参照してください。

ログ記録

boto3 ライブラリは Python の組み込みロギングモジュールと統合されており、セッション中に何が起こるかを追跡できます。ロギングレベルを制御するには、ロギングモジュールを設定します。

import logging

logging.basicConfig(level=logging.INFO)

これにより、INFO 以上のレベルのメッセージを記録するようにルートロガーが設定されます。レベルよりも重大度の低いロギングメッセージは無視されます。ロギングレベルには、DEBUG、INFO、WARNING、ERROR、および CRITICAL があります。デフォルトは WARNING です。

boto3 のロガーは階層構造になっています。ライブラリはいくつかの異なるロガーを使用しており、それぞれがライブラリの異なる部分に対応しています。それぞれの動作を個別に制御できます。

他のライブラリもログに記録されます。内部的には、boto3 は HTTP 接続処理にサードパーティの urllib3 を使用しています。レイテンシーが重要な場合は、ログを監視して、urllib3 が新しい接続を確立したり、アイドル状態の接続を閉じたりするタイミングを確認することで、プールが十分に利用されているかどうかを確認できます。

以下のコードスニペットは、ほとんどのロギングを 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)

イベントフック

Botocore は実行中のさまざまな段階でイベントを発生させます。これらのイベントのハンドラーを登録すると、イベントが発生するたびにハンドラーが呼び出されます。これにより、内部を変更しなくても botocore の動作を拡張できます。

例えば、アプリケーションの DynamoDB テーブルで PutItem 操作が呼び出されるたびに記録しておきたいとしましょう。'provide-client-params.dynamodb.PutItem' イベントに登録すると、関連する Session で PutItem 操作が呼び出されるたびにキャッチしてログに記録できます。例を示します。

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

ハンドラー内で、パラメータをプログラムで操作して動作を変更することもできます。

params['TableName'] = "NewTableName"

イベントについて詳しくは、イベントに関する botocore ドキュメントと、イベントに関する boto3 ドキュメントを参照してください。

ページネーションとページネーター

Query や Scan などの一部のリクエストでは、1 回のリクエストで返されるデータのサイズが制限され、後続のページを取得するにはリクエストを繰り返し行う必要があります。

各ページで読み取られる項目の最大数は、limit パラメータを使用して制御できます。例えば、最後の 10 項目だけが必要な場合、limit を使用して最後の 10 項目だけを取得できます。制限は、フィルタリングが適用される前にテーブルから読み取られる量であることに注意してください。フィルタリング後に正確に 10 を指定する方法はありません。事前にフィルター処理される回数を制御し、実際に 10 個を取得した時点でクライアント側で確認することしかできません。制限にかかわらず、すべての応答の最大サイズは常に 1 MB です。

応答に LastEvaluatedKey が含まれている場合、それは件数またはサイズの制限に達したために応答が終了したことを示します。キーはレスポンスで評価された最後のキーです。この LastEvaluatedKey を取得して，フォローアップコールに ExclusiveStartKey として渡すと、その開始点から次のチャンクを読み取ることができます。LastEvaluatedKey が返されないということは、Query または Scan に一致するアイテムがもうないということです。

以下は、1 ページあたり最大 100 個の項目を読み込み、すべての項目が読み込まれるまでループする単純な例です (Resource インターフェイスを使用していますが、Client インターフェイスでも同じパターンです)。

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

便宜上、boto3 はページネーターを使ってこれを行うことができます。ただし、Client インターフェイスでのみ機能します。以下はページネーターを使うように書き直されたコードです。

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)

詳細については、ページネーターに関するガイドと DynamoDB.Paginator.Query の API リファレンスを参照してください。

ウェイター

ウェイターは、処理が完了するのを待ってから次の処理を進めることができます。現在のところ、サポートしているのはテーブルが作成または削除されるのを待つことだけです。バックグラウンドでは、ウェイター操作は 20 秒ごとに最大 25 回までチェックを行います。これは自分で行うこともできますが、オートメーションを書くときはウェイターを使うのが便利です。

次のコードは、特定のテーブルが作成されるまで待機する方法を示しています。

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

詳細については、ウェイターに関するガイドとウェイターに関するリファレンスを参照してください。