LambdaPowertoolsでDynamoDBからアイテムを取得する
こんにちはインフラ好きなエンジニアのししぃです。
今回は自分の過去記事の棚卸しです。というのも過去に書きかけて放置された記事がいくつもあって、それを見てたら公開ぎりぎりまで書いているけど公開してないものがいくつかあったので、それに少し手を加えて公開します。
ということでこれは2021/11/10に書いて塩漬けにした記事です。
ちなみに余談で、このブログは「動くプロジェクトを公開する」をコンセプトに記事を書いています。ですが、この記事のプロジェクトをどうやら私消しちゃったみたいで、リポジトリは作り直しています。
とか思ってたのですが、2 年前のコードだとほぼほぼ動かない。とくにmotoがこの２年で破壊的変更が何回か入っていてコード自体洗替しました 😥
LambdaPowertools で DynamoDB からアイテムを取得する
今回 DynamoDB のテーブルに入っている情報をLambdaPowertoolsのDynamoDBProviderを使って取得する方法を紹介します。公式資料はこの辺です。
わりと簡単な動作確認になりますので、AWS には接続はせずに moto のモックを使って動作確認を行います。
では 作っていきましょう 👻
まず必要なパッケージをインストールしてきます。
下記のようなファイルをつくって$ pip install -r requirements.txtしてください。
aws-lambda-powertools==2.33.1
boto3==1.34.44
moto==5.0.1
次に動作確認用のコードを書きます。
import boto3
from aws_lambda_powertools.utilities import parameters
from moto import mock_aws
# テーブルを作成する
def create_table(dynamodb, table_name: str):
table = dynamodb.create_table(
TableName=table_name,
KeySchema=[
{"AttributeName": "pk", "KeyType": "HASH"}, # パーティションキー
{"AttributeName": "sk", "KeyType": "RANGE"}, # ソートキー
],
AttributeDefinitions=[
{"AttributeName": "pk", "AttributeType": "S"},
{"AttributeName": "sk", "AttributeType": "S"},
],
ProvisionedThroughput={"ReadCapacityUnits": 1, "WriteCapacityUnits": 1},
)
return table
# テストデータを投入する
def put_test_data(table, test_data):
for item in test_data:
table.put_item(Item=item)
def main():
table_name = "user_config_table"
test_data = [
{"pk": "User_A", "sk": "param_1", "value": "UserAのパラメーター１だよ"},
{"pk": "User_A", "sk": "param_2", "value": "UserAのパラメーター２だよ"},
{"pk": "User_B", "sk": "param_1", "value": "UserBのパラメーター１だよ"},
]
with mock_aws():
dynamodb = boto3.resource("dynamodb")
table = create_table(dynamodb, table_name)
put_test_data(table, test_data)
dynamodb_provider = parameters.DynamoDBProvider(
table_name=table_name, key_attr="pk", sort_attr="sk", value_attr="value"
)
print("########## User_Aのパラメーターを取得")
user_configs = dynamodb_provider.get_multiple("User_A")
for parameter, value in user_configs.items():
print(f"{parameter}: {value}")
print("########## User_Bのパラメーターを取得")
user_configs = dynamodb_provider.get_multiple("User_B")
for parameter, value in user_configs.items():
print(f"{parameter}: {value}")
if __name__ == "__main__":
main()
このコードは、user_config_tableというテーブルを作成し、そのテーブルにUser_AとUser_Bのパラメーターを投入し、それを取得するコードです。DynamoDB のテーブルはモックされているので、実際に AWS アカウントへ接続しなくても動作確認ができます。
これを下記のコマンドで実行してみましょう。
python app.py
下記の出力を得られるはずです。
########## User_Aのパラメーターを取得
param_1: UserAのパラメーター１だよ
param_2: UserAのパラメーター２だよ
########## User_Bのパラメーターを取得
param_1: UserBのパラメーター１だよ
無事 pk が User_A、User_B で登録された、それぞれのパラメーターを取得できることを確認しました。
最後におまけですが、もちろん同様の処理を boto3 を使っても実現できます。それが下記になります。
# (おまけ)boto3を使って直接DynamoDBにアクセスする場合
print("########## User_Aのパラメーターを取得")
response = table.query(
KeyConditionExpression="pk = :pk_value",
ExpressionAttributeValues={
":pk_value": "User_A",
},
)
items = response.get("Items")
for item in items:
parameter = item["sk"]
value = item["value"]
print(f"{parameter}: {value}")
まぁこっちでも全然ありかな？という感じですがLambdaPowertoolsのほうがシンプルに書けるので、プロジェクトにLambdaPowertoolsが入っているなら使ってみるといいかもしれません（あとは今回紹介してませんがキャッシュ機能とかがあって、それをつかって通信回数を減らしたいとかの要件がある場合も）。逆に Dynamo からデータを取得したいという要件だけのためにLambdaPowertoolsを入れるという選択はちょっとないかな、、、という感じですね。
はい。以上です、ありがとうございました！！
今回のリポジトリはこちら
https://github.com/sisi100/powertools-dynamodbprovider-sample