タイトル : DynamoDB テストデータ作成 localstackを試そう 2024
更新日 : 2025-01-03
カテゴリ : プログラミング
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万件のデータを用意出来ました