Python AWS DynamoDBのシリアライザー・デシリアライザー

どうも、nippa です。

AWS DynamoDB で DynamoDB の専用の型を dict 型を渡す必要があったり、データを受け 取ったあとに処理しないといけません。

boto3 の dynamodb のライブラリの中には、その処理を簡略化する serializer、deserializer が用意されています。

今回はこの 2 つの使い方について記事にします。

環境

• macOS 12.6
• python 3.9.16
• boto3 1.26.155

シリアライズ（dict -> DynamoDB 形式）

DynamoDB にデータを put_item する場合、通常 DynamoDB の型を記載する必要がありま す。

以下のようになります。

def put_item() -> None:
    table_name = "test"
    response = dynamodb.put_item(
        TableName=table_name,
        Item={
            "id": {"N": "1"},
            "email": {"S": "example@example.com"},
            "name": {"S": "Yamada"},
            "age": {"N": "30"},
        },
    )
    print(response)
    return

DynamdDB の型を Boto3 で準備されているシリアライザーを利用することで、簡略化する ことができます。

シリアライザーで書き直したものが以下になり、適切な形に書き換えてくれます。

import boto3
from boto3.dynamodb.types import TypeSerializer


dynamodb = boto3.client("dynamodb", region_name="us-west-2")
serializer = TypeSerializer()

def create_table() -> None:
    table_name = "test"
    dynamodb.create_table(
        TableName=table_name,
        KeySchema=[
            {"AttributeName": "id", "KeyType": "HASH"},  # Partition key
            {"AttributeName": "email", "KeyType": "RANGE"},  # Sort key
        ],
        AttributeDefinitions=[
            {"AttributeName": "id", "AttributeType": "N"},
            {"AttributeName": "email", "AttributeType": "S"},
        ],
        BillingMode="PAY_PER_REQUEST",
    )

    # table作成完了まで待つ
    dynamodb.get_waiter("table_exists").wait(TableName=table_name)

    print(f"Table {table_name} created successfully.")
    return


def put_item() -> None:
    table_name = "test"
    item: dict = {"id": 1, "email": "example@example.com", "name": "Yamada", "age": 30}

    response = dynamodb.put_item(
        TableName=table_name,
        Item={k: serializer.serialize(v) for k, v in item.items()},
    )
    print(response)
    return

if __name__ == "__main__":
    create_table()
    put_item()

デシリアライズ（DynamoDB 形式 -> dict）

一般的な DynamoDB からアイテムを取得した場合のレスポンスは以下のようになります。

def get_item() -> None:
    table_name = "test"
    response = dynamodb.get_item(
        TableName=table_name,
        Key={
            "id": {"N": "1"},
            "email": {"S": "example@example.com"},
        },
    )
    item = response.get("Item")
    if item:
        print(item)
    else:
        print("Item not found")
    return

# response
{'id': {'N': '1'}, 'email': {'S': 'example@example.com'}, 'name': {'S': 'Yamada'}, 'age': {'N': '30'}}

この形では、取り扱うための処理が必要になりますが、Boto3 で用意されているデシリア ライザーを利用すると、取り扱いが非常に楽になります。

デシリアライザーを利用した場合は、以下のようになります。

import boto3
from boto3.dynamodb.types import TypeDeserializer

dynamodb = boto3.client("dynamodb", region_name="us-west-2")
deserializer = TypeDeserializer()

def get_item() -> None:
    table_name = "test"
    response = dynamodb.get_item(
        TableName=table_name,
        Key={
            "id": {"N": "1"},
            "email": {"S": "example@example.com"},
        },
    )
    item = response.get("Item")
    if item:
        print({k: deserializer.deserialize(v) for k, v in item.items()})
    else:
        print("Item not found")
    return


if __name__ == "__main__":
    get_item()

# response
{'id': Decimal('1'), 'email': 'example@example.com', 'name': 'Yamada', 'age': Decimal('30')}

感想

今回は DynamoDB のシリアライザー、デシリアライザーの使い方についてまとめました。

データの取り扱いが非常に便利になりますになりますので、是非活用してみてください。

ではでは、また次回。