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

AWS localstackを試そう 2024

DynamoDBでクエリーしましょう

SCANは全検索になるのでキャパシティに影響するらしい。GSIを作ればクエリーで書けるってことね。

DynamoDB基礎知識とboto3での簡単な使用方法

ソース

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

# クライアントの用意
dd_resource = boto3.resource("dynamodb")
table = dd_resource.Table("ProcTable")

# クエリー
#   パーティションキー PDate が 20250102 である
res_query = table.query(
    KeyConditionExpression=Key("PDate").eq("20250102"),
)
# クエリーで見つかった個数
count = int(res_query["Count"])
print(f"QUERY COUNT : {count}")

# 見つかったItemsでループ
err_count = 0
for item in res_query["Items"]:
    # 処理内容を取得
    proc_json = item["ProcContent"]
    # 処理内容がNGのものを表示
    if proc_json["status"] == "NG":
        # print(item)
        err_count += 1
print(f"ERROR COUNT : {err_count}")

# GSIのShopIndexでクエリー
# 店舗 : shop01 の 店員 : shop01_c010 を条件にする
res_query = table.query(
    IndexName="ShopIndex",
    KeyConditionExpression=Key("ShopId").eq("shop01")
    & Key("ClerkId").eq("shop01_c010"),
)
# クエリーで見つかった個数
count = int(res_query["Count"])
print(f"QUERY COUNT : {count}")

# GSIでクエリー出来るけど、スキャンでやってみる
res_scan = table.scan(
    FilterExpression=Attr("ClerkId").contains("shop01_c010"),
)
# クエリーで見つかった個数
count = int(res_scan["Count"])
print(f"SCAN COUNT : {count}")

実行

$ python test_query_01.py 
QUERY COUNT : 1000
ERROR COUNT : 53
QUERY COUNT : 242
SCAN COUNT : 242
$