トップ > Amazon DynamoDB > AWS SDK for Python (Boto3)を使って、DynamoDBを操作する

AWS SDK for Python (Boto3)を使って、DynamoDBを操作する

Amazon DynamoDB boto3
記事タイトルとURLをコピーする

こんにちは。AWS CLIが好きな福島です。

今回は、AWS SDK for Python (Boto3)を使って、DynamoDBを操作してみたいと思います。

参考情報

テーブルの作成(create_table)

テーブルを作成するためには、create_tableメソッドを利用します。

create_table.py

import boto3

TABLE_NAME = "todo-table"

dynamodb_client = boto3.client("dynamodb")


dynamodb_client.create_table(
    TableName=TABLE_NAME,
    AttributeDefinitions=[
        {"AttributeName": "UserID", "AttributeType": "S"},
        {"AttributeName": "TaskID", "AttributeType": "S"},
    ],
    KeySchema=[
        {"AttributeName": "UserID", "KeyType": "HASH"},
        {"AttributeName": "TaskID", "KeyType": "RANGE"},
    ],
    ProvisionedThroughput={"ReadCapacityUnits": 10, "WriteCapacityUnits": 10},
)

指定している引数は以下の通りです。

  • TableName:
    • 操作するテーブル名を指定します。
  • AttributeDefinitions:
    • 属性を定義します。
  • KeySchema:
    • 定義した属性のキータイプを指定します。
      • HASH: パーティションキー
      • RANG: ソートキー
  • ProvisionedThroughput:
    • Read,Writeのキャパシティを設定します。

※ソートキーの設定は任意です。

アイテムの追加(put_item)

アイテムを追加するためには、put_itemメソッドを利用します。

put_item.py

import boto3

TABLE_NAME = "todo-table"

dynamodb_client = boto3.client("dynamodb")

response = dynamodb_client.put_item(
    TableName=TABLE_NAME,
    Item={
        "UserID": {"S": "tanaka"},
        "TaskID": {"S": "1"},
        "Title": {"S": "Learn DynamoDB"},
    },
)

指定している引数は以下の通りです。

  • TableName:
    • 操作するテーブル名を指定します。
  • Item:
    • 追加するアイテムを指定します。
    • 追加するアイテムには、パーティションキー及びソートキー(設定している場合)が必要となります。
    • 今回は、UserID, TaskID, Titleを追加しています。

アイテムの取得(get_item)

特定のアイテムを取得するためには、get_itemメソッドを利用します。このメソッドは、パーティションキーとソートキー(設定している場合)を使ってアイテムを取得します。

get_item.py

import boto3

TABLE_NAME = "todo-table"

dynamodb_client = boto3.client("dynamodb")

response = dynamodb_client.get_item(
    TableName=TABLE_NAME,
    Key={
        "UserID": {"S": "tanaka"},
        "TaskID": {"S": "1"},
    }
)

print(response.get("Item"))

指定している引数は以下の通りです。

  • TableName:
    • 操作するテーブル名を指定します。
  • Key:
    • 取得するアイテムのパーティション及びソートキー(設定している場合)を指定します。

実行すると以下のような出力が得られると思います。

$ python get_item.py
{'Title': {'S': 'Learn DynamoDB'}, 'UserID': {'S': 'tanaka'}, 'TaskID': {'S': '1'}}
$

アイテムの更新(update_item)

既存のアイテムを更新するためには、update_itemメソッドを利用します。

update_item.py

import boto3

TABLE_NAME = "todo-table"

dynamodb_client = boto3.client("dynamodb")

response = dynamodb_client.update_item(
    TableName=TABLE_NAME,
    Key={
        "UserID": {"S": "tanaka"},
        "TaskID": {"S": "1"},
    },
    UpdateExpression="SET Title = :new_title",
    ExpressionAttributeValues={
        ":new_title": {"S": "Update DynamoDB"},
    },
)

指定している引数は以下の通りです。

  • TableName:
    • 操作するテーブル名を指定します。
  • Key:
    • 更新するアイテムのパーティション及びソートキー(設定している場合)を指定します。
  • UpdateExpression:
    • アイテムを更新するための式を設定します。
    • SET [アイテム名] = :[プレースホルダ(任意の値)]のように記述します。
    • 今回は、Title属性を更新するためにSET Title = :new_titleと設定しています。
  • ExpressionAttributeValues:
    • 更新したい値を設定します。
    • ":[プレースホルダ]": [更新データ]のように記述します。
    • 今回は、TitleをUpdate DynamoDBに更新するため、":new_title": {"S": "Update DynamoDB"}としています。

※プレースホルダは任意の値ですが、先頭の:は省略できず、UpdateExpressionとExpressionAttributeValuesで指定するプレースホルダは一致している必要があります。

: 記号を省略することはできません。

アイテムを更新後、get_itemメソッドでアイテムを確認すると、以下のようにTitleがUpdate DynamoDBに更新されていることが分かると思います。

$ python get_item.py
{'Title': {'S': 'Update DynamoDB'}, 'UserID': {'S': 'tanaka'}, 'TaskID': {'S': '1'}}
$

補足

複数のキーを更新したい場合は、UpdateExpressionの値は、,で区切り値を追加します。 ExpressionAttributeValuesには、要素を追加すればOKです。

response = dynamodb_client.update_item(
    TableName=TABLE_NAME,
    Key={
        "UserID": {"S": "tanaka"},
        "TaskID": {"S": "1"},
    },
    UpdateExpression="SET Title = :new_title, Description = :new_description",
    ExpressionAttributeValues={
        ":new_title": {"S": "Update DynamoDB"},
        ":new_description": {"S": "Update multiple attributes in DynamoDB"},
    },
)

アイテムの削除(delete_item)

アイテムを削除するためには、delete_itemメソッドを利用します。このメソッドは、指定したパーティションキー及びソートキー(設定している場合)に基づいてアイテムを削除します。

delete_item.py

import boto3

TABLE_NAME = "todo-table"

dynamodb_client = boto3.client("dynamodb")

response = dynamodb_client.delete_item(
    TableName=TABLE_NAME,
    Key={
        "UserID": {"S": "tanaka"},
        "TaskID": {"S": "1"},
    }
)

指定している引数は以下の通りです。

  • TableName:
    • 操作するテーブル名を指定します。
  • Key:
    • 削除するアイテムのパーティション及びソートキー(設定している場合)を指定します。

アイテムを削除後、get_itemメソッドでアイテムを確認すると、以下のようにNone(アイテムが削除されている)ことが分かると思います。

$ python get_item.py
None
$

複数アイテムの一括追加(batch_write_item)

DynamoDBに複数のアイテムを一度に追加する場合、batch_write_itemメソッドを使用します。

batch_write_item.py

import boto3

TABLE_NAME = "todo-table"

dynamodb_client = boto3.client("dynamodb")

# バッチで追加するアイテムのリスト
items_to_add = [
    {
        "PutRequest": {
            "Item": {
                "UserID": {"S": "tanaka"},
                "TaskID": {"S": "1"},
                "Title": {"S": "Learn DynamoDB"},
            }
        }
    },
    {
        "PutRequest": {
            "Item": {
                "UserID": {"S": "tanaka"},
                "TaskID": {"S": "2"},
                "Title": {"S": "Learn Python"},
            }
        }
    },
    {
        "PutRequest": {
            "Item": {
                "UserID": {"S": "tanaka"},
                "TaskID": {"S": "3"},
                "Title": {"S": "Write Blog Post"},
            }
        }
    },
    {
        "PutRequest": {
            "Item": {
                "UserID": {"S": "ueda"},
                "TaskID": {"S": "1"},
                "Title": {"S": "Write Blog Post"},
            }
        }
    }
]

# バッチ書き込みの実行
response = dynamodb_client.batch_write_item(
    RequestItems={
        TABLE_NAME: items_to_add
    }
)

指定している引数は以下の通りです。

  • TableName:

    • 操作するテーブル名を指定します。

  • RequestItems:

    • このパラメータは、テーブル名をキーとする辞書で、その値として追加するリクエストのリストを指定します。ここでは、PutRequestを使ってアイテムを追加します。

  • PutRequest:

    • 各アイテムのデータを定義するためのリクエストになります。Itemキーの下に属性とその値を指定します。

クエリ操作(query)

DynamoDBから条件に合致したアイテムを取得するために、queryメソッドを利用できます。 今回は、UserIDがtanakaのアイテムを取得します。

query.py

import boto3

TABLE_NAME = "todo-table"

dynamodb_client = boto3.client("dynamodb")

response = dynamodb_client.query(
    TableName=TABLE_NAME,
    KeyConditionExpression="UserID = :userid",
    ExpressionAttributeValues={
        ":userid": {"S": "tanaka"}
    }
)

items = response.get("Items", [])
for item in items:
    print(item)

指定している引数は以下の通りです。

  • TableName:
    • 操作するテーブル名を指定します。
  • KeyConditionExpression:
    • クエリの条件式を設定します。
    • [アイテム名] = :[プレースホルダ(任意の値)]のように記述します。
    • 今回は、UserIDキーの値を検索するため、UserID = :useridと設定しています。
  • ExpressionAttributeValues:
    • 検索に利用する値を設定します。
    • ":[プレースホルダ]": [検索したい値]のように記述します。
    • 今回は、UserIDがtanakaに等しいアイテムを検索しています。

実行すると以下のように複数のアイテムを取得できます。

$ python query.py
{'Title': {'S': 'Learn DynamoDB'}, 'UserID': {'S': 'tanaka'}, 'TaskID': {'S': '1'}}
{'Title': {'S': 'Learn Python'}, 'UserID': {'S': 'tanaka'}, 'TaskID': {'S': '2'}}
{'Title': {'S': 'Write Blog Post'}, 'UserID': {'S': 'tanaka'}, 'TaskID': {'S': '3'}}
$

Scan操作(scan)

条件を指定せずにDynamoDBから複数のデータを取得したい場合は、scanメソッドを利用できます。

scan.py

import boto3

TABLE_NAME = "todo-table"

dynamodb_client = boto3.client("dynamodb")

response = dynamodb_client.scan(
    TableName=TABLE_NAME,
)

items = response.get("Items", [])
for item in items:
    print(item)

指定している引数は以下の通りです。

  • TableName:
    • 操作するテーブル名を指定します。

実行すると以下のように複数のアイテムを取得できます。

$ python scan.py 
{'Title': {'S': 'Learn DynamoDB'}, 'UserID': {'S': 'tanaka'}, 'TaskID': {'S': '1'}}
{'Title': {'S': 'Learn Python'}, 'UserID': {'S': 'tanaka'}, 'TaskID': {'S': '2'}}
{'Title': {'S': 'Write Blog Post'}, 'UserID': {'S': 'tanaka'}, 'TaskID': {'S': '3'}}
{'Title': {'S': 'Write Blog Post'}, 'UserID': {'S': 'ueda'}, 'TaskID': {'S': '1'}}
$

1M以上のデータ取得

queryとscanメソッドは一度のリクエストで取得できるデータが1Mに制限されます。

1 回の Query オペレーションで、最大 1 MB のデータを取得できます。

1 回の Scan リクエストで、最大 1 MB のデータを取得できます。

1M以上のデータを取得するサンプルコード

ポイントとして1MB以上のデータがある場合、QueryもScanもレスポンスにLastEvaluatedKeyが含まれます。 そのため、LastEvaluatedKeyが存在しなくなるまでループする処理を実装することで1MB以上のデータを取得できます。

Queryで1M以上のデータを取得するサンプルコード

all_query.py

import boto3

TABLE_NAME = "todo-table"

dynamodb_client = boto3.client("dynamodb")


def query_items(user_id):
    items = []
    last_evaluated_key = None

    query_kwargs = {
        "TableName": TABLE_NAME,
        "KeyConditionExpression": "UserID = :userid",
        "ExpressionAttributeValues": {":userid": {"S": user_id}},
    }

    while True:

        if last_evaluated_key:
            query_kwargs["ExclusiveStartKey"] = last_evaluated_key

        # クエリの実行
        response = dynamodb_client.query(**query_kwargs)

        # 取得したアイテムをリストに追加
        items.extend(response.get("Items", []))

        # LastEvaluatedKeyが存在する場合は、次のセットを取得
        last_evaluated_key = response.get("LastEvaluatedKey")

        # LastEvaluatedKeyが存在しない場合、すべてのアイテムが取得済み
        if not last_evaluated_key:
            break

    return items


# 使用例: UserID = "tanaka"のアイテムをクエリ
all_items_query = query_items("tanaka")

for item in all_items_query:
    print(item)

Scanで1M以上のデータを取得するサンプルコード

all_scan.py

import boto3

TABLE_NAME = "todo-table"

dynamodb_client = boto3.client("dynamodb")


def scan_items():
    items = []
    last_evaluated_key = None
    scan_kwargs = {"TableName": TABLE_NAME}

    while True:

        if last_evaluated_key:
            scan_kwargs["ExclusiveStartKey"] = last_evaluated_key

        # スキャンの実行
        response = dynamodb_client.scan(**scan_kwargs)

        # 取得したアイテムをリストに追加
        items.extend(response.get("Items", []))

        # LastEvaluatedKeyが存在する場合は、次のセットを取得
        last_evaluated_key = response.get("LastEvaluatedKey")

        # LastEvaluatedKeyが存在しない場合、すべてのアイテムが取得済み
        if not last_evaluated_key:
            break

    return items


# 使用例: テーブルの全アイテムをスキャン
all_items_scan = scan_items()

for item in all_items_scan:
    print(item)

動作確認

まずは1MB以上のデータを登録します。 今回は、1アイテム、約400KBのデータを5個追加します。

add_large_data.py

import boto3

# DynamoDBのクライアントを作成
dynamodb_client = boto3.client("dynamodb")

TABLE_NAME = "todo-table"


# データをDynamoDBに追加
def add_large_data(user_id):
    data = "a" * 390 * 1024
    for i in range(5):
        print(i)
        dynamodb_client.put_item(
            TableName=TABLE_NAME,
            Item={
                "UserID": {"S": user_id},
                "TaskID": {"S": str(i)},
                "Title": {"S": f"{i} {data}"},
            },
        )


add_large_data("tanaka")

それぞれのサンプルコードを実行すると、1MB以上のデータが取得できていることを確認できると思います。

$ python query.py | wc -l
       3
$ python all_query.py | wc -l
       5
$ 

$ python scan.py | wc -l
       3
$ python all_scan.py | wc -l
       6
$

whileループ処理を入れていないquery.py, scan.pyは、3件(一部)のデータだけを取得し、 whileループ処理を入れているall_query.py, all_scan.pyは、全部データを取得できていることが分かります。

all_query.pyとall_scan.pyの件数が違うのは、UserIDがtanakaでフィルターをかけているか、かけていないかの違いです。

まとめ

今回は、AWS SDK for Python (Boto3)を使って、DynamoDBを操作するブログをまとめました。 どなたかのお役に立てれば幸いです。

福島 和弥 (記事一覧)

2019/10 入社

AWS CLIが好きです。

【お知らせ】

【AWS re:Invent 2024 をエンジニアブログで振り返ろう】

2024年12月2日 - 12月6日 | ラスベガス（米国）
世界最大の AWS カンファレンス・AWS re:Invent 2024の感動をエンジニアブログで振り返りましょう！

AWS re:Invent 2024のブログ記事を読む