Amazon DynamoDB Local で新たに2つの DynamoDB API をサポートしました

こんにちは、森田です。

以下のアップデートで Amazon DynamoDB Local にて新しい DynamoDB API を利用することができるようになりました。

今回新たにサポートされたAPIは以下の通りです。

• deletion protection（削除保護）
• ReturnValuesOnConditionCheckFailure（条件付き書き込み失敗時のレスポンス設定）

これら2つの機能は、DynamoDBで2023年に発表されたものですが、今まで DynamoDB Local では対応しておらず、テストを行うことができませんでした。

今回のアップデートで、これらのAPIに対しても対応できるようになったため、DynamoDB Local でも動作テストができるようになります。

やってみた

では、実際に Dynamo DB Local を利用して新しいAPIを叩いてみます。

DynamoDB Local の起動

本記事では、dynamodb-local イメージを利用して DynamoDB Local をコンテナ上で起動させます。

docker pull amazon/dynamodb-local
docker run -d --name dynamodb-local -p 8000:8000 amazon/dynamodb-local

削除保護

今回は、boto3を利用してAPIを叩いてみます。

削除保護を有効にしてテーブルの作成

以下のコードを実行して、テーブルの作成を行います。

DynamoDB Local を利用するために、endpoint_urlを指定します。

create.py

import boto3

# DynamoDBリソースを作成し、エンドポイントURLを指定します。
dynamodb = boto3.resource('dynamodb', endpoint_url='http://localhost:8000/')

# テーブルを作成するためのパラメータを設定します。
table_name = 'User'
attribute_definitions = [
    {
        'AttributeName': 'userId',
        'AttributeType': 'S' 
    }
]
key_schema = [
    {
        'AttributeName': 'userId',
        'KeyType': 'HASH'  
    }
]
provisioned_throughput = {
    'ReadCapacityUnits': 5,
    'WriteCapacityUnits': 5
}

# テーブルを作成します。
table = dynamodb.create_table(
    TableName=table_name,
    AttributeDefinitions=attribute_definitions,
    KeySchema=key_schema,
    ProvisionedThroughput=provisioned_throughput,
    DeletionProtectionEnabled=True #削除保護
)

# テーブルの作成が完了するまで待ちます。
table.meta.client.get_waiter('table_exists').wait(TableName=table_name)

print(f"Table {table_name} has been created.")

実行結果

$ python3 create.py
Table User has been created.

削除保護を有効にしたテーブルの削除

以下のコードを実行しても、削除することができず、エラーが発生します。

delete.py

import boto3

table_name = "User"

# DynamoDBクライアントを作成します。
dynamodb = boto3.client('dynamodb', endpoint_url='http://localhost:8000/')

dynamodb.delete_table(TableName=table_name)

実行結果

$ python3 delete.py
botocore.exceptions.ClientError: An error occurred (ValidationException) when calling the DeleteTable operation: Resource cannot be deleted as it is currently protected against deletion. Disable deletion protection first.

削除保護を無効後、テーブルの削除

では、削除できるように、削除保護を無効化して、テーブルの削除を行います。

delete2.py

import boto3

table_name = "User"

# DynamoDBクライアントを作成します。
dynamodb = boto3.client('dynamodb', endpoint_url='http://localhost:8000/')

# 削除保護を無効化
response = dynamodb.update_table(
    TableName=table_name,
   DeletionProtectionEnabled=False #削除保護
)

dynamodb.delete_table(TableName=table_name)

実行結果

$ python3 delete2.py

今度はエラーなく実行ができ、テーブルの削除が行えます。

ReturnValuesOnConditionCheckFailure

今回は以下の記事のコードを参考に動作検証を行います。

テーブル作成とアイテムの追加

以下のコードで動作確認用のテーブル作成とアイテム追加を行います。

item_create.py

import boto3

dynamodb = boto3.client("dynamodb",  endpoint_url='http://localhost:8000/')

table_name = "sample_table"

response = dynamodb.create_table(
    TableName=table_name,
    KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
    AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "N"}],
    BillingMode="PAY_PER_REQUEST",
)

dynamodb.get_waiter("table_exists").wait(TableName=table_name)
print("テーブル 作成成功")

item = {
    "id": {"N": "1"},
    "name": {"S": "yamada taro"},
    "email": {"S": "yamada.taro@example.com"},
}

dynamodb.put_item(
    TableName=table_name,
    Item=item,
)

print("アイテム 作成成功")

実行結果

$ python3 item_create.py 
テーブル 作成成功
アイテム 作成成功

条件付き書き込み

続いて以下のコードを実行して、条件付き書き込みを行います。

idが存在しない場合に書き込みが成功する指定を行っていますが、先ほどid=1のアイテムを追加しているためエラーとなります。

put.py

import boto3
from botocore.exceptions import ClientError

dynamodb = boto3.client("dynamodb", endpoint_url='http://localhost:8000/')
table_name = "sample_table"

# idのみが同一のitem
item = {
    "id": {"N": "1"},
    "name": {"S": "satou hanako"},
    "email": {"S": "satou.hanako@example.com"},
}

try:
    dynamodb.put_item(
        TableName=table_name,
        Item=item,
        ConditionExpression="attribute_not_exists(id)",  # idが存在しない場合にのみ書き込み
        ReturnValuesOnConditionCheckFailure="ALL_OLD",
    )
    print("Item 追加成功")
except ClientError as e:
    if e.response["Error"]["Code"] == "ConditionalCheckFailedException":
        print(e.response.get("Item"))

実行結果

$ python3 put.py
{'name': {'S': 'yamada taro'}, 'email': {'S': 'yamada.taro@example.com'}, 'id': {'N': '1'}}

実行結果からも、期待した通りに、元のアイテムをエラー時に返却しています。

さいごに

今回新しいAPIにも対応したため、ますます DynamoDB Local が使いやすくなりました。

DynamoDB　をテスト用で利用したい場合には、 DynamoDB Local は非常に便利なツールですので、ぜひチェックしてみてください！