【AWS】Lambda(boto3)を使ってDynamoDBへ読み/書き込みを実施

はじめに

Lambdaを使用してDynamoDBのテーブル読み込みをできますか？
私はSDKを利用してdynamodb.queryなどの関数を使用すればいいと考えていましたが、具体的な実装まで理解できていませんでした。

本記事はLambdaを利用してDynamoDBに対して、
一般的なアクションを実装してみた記事となります。

今回試してみる構成

This is Simple構成

DynamoDB構成

パーティションキー：Id
ソートキー：Timestanp

前提

クライアントインターフェイス

低レベルなインターフェースであり抽象化は最小限。
利用者側で細かな制御を行いDynamoDBを操作することができる。
AWS API にそのままリクエストするような形で呼び出せるようになっている。

リソースインターフェイス

高レベルばインターフェースであり抽象化がされている。
利用者側で細かな制御は不要でDynamoDBを操作することができる。
オブジェクト指向のようにメソッドを利用して書けるよう用意された高レベルなライブラリとなる。

実装してみた

DynamoDBからデータを読み込み(query)

以下2パターンのインターフェースを用いて書き込みアクションを実装

クライアントインターフェースを用いた実装

記載の条件式についてはBoto3ドキュメント内KeyConditionExpression参照
今回の取得対象は以下に一致する項目をテーブルから取得する。

Id (S)	Timestanp (N)	Title
user_111	20240629	test1

import boto3
import logging
import traceback
# from boto3.dynamodb.conditions import Key, Attr

# モジュールごと(pythonファイルごと)にログ出力するloggerインスタンス作成
logger = logging.getLogger(__name__)
# ログレベル[INFO]を設定
logger.setLevel(logging.INFO)
############################
## Lambdaメイン処理
############################
def lambda_handler(event, context):
    try:
        logger.info("start_lambda")
        # クライアントインターフェースを指定
        dynamodb = boto3.client('dynamodb')
        response = dynamodb.query(
            # 「TableName」：記載必須項目であるDynamoDBのテーブル名
            TableName = 'testtable',
            # 「KeyConditionExpression」：パーティションキーとソートキーによる条件式
            # 右辺の式は変数に代入しているわけではなくSQLでいうWhere句のような条件
            # 「パーティションキー：user_111」かつ「ソートキー：Timestanp」が20240629以上である項目を取得
            KeyConditionExpression='Id = :Id_val AND Timestanp >= :Timestanp_val',
            # 「FilterExpression」キー以外の項目に対する条件式
            # 「Title：test1」である項目を取得
            FilterExpression='Title = :Title',
            # 「ExpressionAttributeValues」：クエリやフィルタ式で使用される値の変数を定義する
            ExpressionAttributeValues={
                ':Id_val':{'S':'user_111'},
                ':Timestanp_val':{'N': '20240629' },
                ':Title':{'S': 'test1'}
                }
            )
        logger.info(response)
        return response
    except Exception as e:
        logger.error(e)
        traceback.print_exc()
        raise e

※補足「Timestanp_val':{'N': '20240629' }」数値型を入力する項目に、
''でstr型のようにしている理由は、以下の通りです。

Numbers are sent across the network to DynamoDB as strings, to maximize compatibility across languages and libraries. However, DynamoDB treats them as number type attributes for mathematical operations.
要約すると、数値は言語やライブラリ間の互換性を最大化するために、文字列として設定する必要がある。
参照先：https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb/client/query.html

リソースインターフェースを用いた実装

クライアントインターフェースと違って記述量がかなり減りました。
特に「ExpressionAttributeValues」で明示的に変数化/属性付与が不要になったのは大きな削減。
記載の条件式についてはDynamoDB 条件参照

今回の取得対象は以下に一致する項目をテーブルから取得する。

Id (S)	Timestanp (N)	Title
user_111	20240629	test1

import boto3
import logging
import traceback
from boto3.dynamodb.conditions import Key, Attr

# モジュールごと(pythonファイルごと)にログ出力するloggerインスタンス作成
logger = logging.getLogger(__name__)
# ログレベル[INFO]を設定
logger.setLevel(logging.INFO)
client = boto3.client('dynamodb')
############################
## Lambdaメイン処理
############################
def lambda_handler(event, context):
    try:
        logger.info("start_lambda")
        # リソースインターフェースを指定
        dynamodb = boto3.resource('dynamodb')
        table = dynamodb.Table('testtable')
        response = table.query(
            # 「KeyConditionExpression」：パーティションキーとソートキーによる条件式
            # 「パーティションキー：user_111」かつ「ソートキー：Timestanp」が20240629以上である項目を取得
            KeyConditionExpression=Key('Id').eq("user_111") & Key('Timestanp').gte(20240629),
            # 「FilterExpression」キー以外の項目に対する条件式
            # 「Attr」はDynamoDBのクエリスキャンで使用される属性表現につきパッケージimpotrが必要
            # 「Title：test1」である項目を取得
            FilterExpression=Attr('Title').eq("test1")
            )
        logger.info(response)
        return response
    except Exception as e:
        logger.error(e)
        traceback.print_exc()
        raise e

DynamoDBにデータを書き込み(put)

以下2パターンのインターフェースを用いて読み込みアクションを実装

クライアントインターフェースを用いた実装

今回は以下情報をテーブルに追加する。

Id (S)	Timestanp (N)	Title
user_111	20240631	test3

import boto3
import logging
import traceback
# from boto3.dynamodb.conditions import Key, Attr

# モジュールごと(pythonファイルごと)にログ出力するloggerインスタンス作成
logger = logging.getLogger(__name__)
# ログレベル[INFO]を設定
logger.setLevel(logging.INFO)
############################
## Lambdaメイン処理
############################
def lambda_handler(event, context):
    try:
        logger.info("start_lambda")
        # クライアントインターフェースを指定
        dynamodb = boto3.client('dynamodb')
        response = dynamodb.put_item(
            TableName = 'testtable',
            Item={
                'Id':{'S':'user_111'},
                'Timestanp':{'N': '20240631' },
                'Title':{'S': 'test3'}
                },
            )
        logger.info(response)
        return response
    except Exception as e:
        logger.error(e)
        traceback.print_exc()
        raise e

リソースインターフェースを用いた実装

今回は以下情報をテーブルに追加する。

Id (S)	Timestanp (N)	Title
user_111	20240629	test4

import boto3
import logging
import traceback
from boto3.dynamodb.conditions import Key, Attr

# モジュールごと(pythonファイルごと)にログ出力するloggerインスタンス作成
logger = logging.getLogger(__name__)
# ログレベル[INFO]を設定
logger.setLevel(logging.INFO)
############################
## Lambdaメイン処理
############################
def lambda_handler(event, context):
    try:
        logger.info("start_lambda")
        # リソースインターフェースを指定
        dynamodb = boto3.resource('dynamodb')
        table = dynamodb.Table('testtable')
        response = table.put_item(
            Item={
                'Id':':user_111',
                'Timestanp':20240629,
                'Title':'test4'
                },
            )
        logger.info(response)
        return response
    except Exception as e:
        logger.error(e)
        traceback.print_exc()
        raise e

終わりに

本記事がDynamoDBへのアクション方法で悩んでいる方の助けになれば幸いです。

参考記事