タイトル : DynamoDB テストデータ作成 localstackを試そう 2024
更新日 : 2025-01-03
カテゴリ : プログラミング
タグ :
aws   
python   
boto_3   
dynamodb   

AWS localstackを試そう 2024

DynamoDBでクエリーをやる前にデータを作りましょう

ソース

import random
import sys

import boto3
from boto3.dynamodb.types import TypeSerializer

# 型ヒント
from mypy_boto3_dynamodb import DynamoDBClient

# クライアントの用意
dd_client: DynamoDBClient = boto3.client(
    "dynamodb",
)

# テーブル名
TABLE_NAME = "ProcTable"

# boto3のDynamoDB用のJSON変換シリアライザー用意
serializer = TypeSerializer()

# 日付の初日
date_int = 20250102
# 店舗
shops = [f"shop{s_id:02}" for s_id in range(1, 5)]
# 店員
clerks = [f"c{c_id:03}" for c_id in range(1, 11)]
# 処理タイプ
ptypes = ["GET", "POST", "DELETE", "PUT"]
# STATUS
stats = ["OK", "NG"]
stats_weights = [0.95, 0.05]

for date_index in range(10):
    # 日付index
    p_date = str(date_int + date_index)

    # 1日の中で処理IDはユニーク
    for proc_id_int in range(1000):
        # 処理ID
        proc_id = f"proc{proc_id_int:08}"
        # 店舗ID
        shop_id = random.choices(shops)[0]
        # 店員ID
        clerk_id_tmp = random.choices(clerks)[0]
        clerk_id = f"{shop_id}_{clerk_id_tmp}"
        # 処理内容
        ptype = random.choices(ptypes)[0]
        pstatus = random.choices(stats, stats_weights)[0]
        proc_content = {"type": ptype, "status": pstatus, "messaee": "XXX"}
        # シリアライズ
        proc_content4dd = {k: serializer.serialize(v) for k, v in proc_content.items()}

        # 書き込み
        res_put = dd_client.put_item(
            TableName=TABLE_NAME,
            Item={
                "PDate": {"S": p_date},
                "ProcId": {"S": proc_id},
                "ShopId": {"S": shop_id},
                "ClerkId": {"S": clerk_id},
                "ProcContent": {"M": proc_content4dd},
            },
        )

実行

$ python make_data01.py 
$ aws dynamodb scan --table-name ProcTable --select "COUNT"
{
    "Count": 10000,
    "ScannedCount": 10000,
    "ConsumedCapacity": null
}
$ 

1万件のデータを用意出来ました