概要
なるべく行数や文字数を最小限に抑えつつ、PythonからDynamoDBを操作する方法を網羅する(したい)記事
- 理由
いちいちドキュメント読みたくない←
他の人の記事の中から必要な情報を毎回探し出すのも面倒←←←(ひどい)
実際に試したものから書いていくぞ( *˙︶˙*)وグッ!
操作一覧
全操作に共通する事前準備
import boto3
from boto3.dynamodb.conditions import Attr, Key
# region_name には自分のリージョンの名前を書いてね
dynamodb = boto3.resource('dynamodb', region_name='us-east-2')
1. create_table() ... テーブル作成
テーブル作成
table_name = 'hoge_table'
table = dynamodb.create_table(
TableName = table_name,
KeySchema = [
{'AttributeName': 'primary_key_name', 'KeyType': 'HASH'},
{'AttributeName': 'column_name', 'KeyType': 'RANGE'}
],
AttributeDefinitions = [
{'AttributeName': 'primary_key_name', 'AttributeType': 'N'},
{'AttributeName': 'column_name', 'AttributeType': 'S'}
],
ProvisionedThroughput = {
'ReadCapacityUnits': 1, 'WriteCapacityUnits': 1
}
)
table.meta.client.get_waiter('table_exists').wait(TableName=table_name)
- KeySchema
-
AttributeName
でカラム名を指定
'primary_key_name'や'column_name'はあくまでもサンプルの名前 >_<; -
'KeyType': 'HASH'
で、パーティションキーになる -
'KeyType': 'RANGE'
で、ソートキーになる
-
- AttributeDefinitions
-
'AttributeType': 'S'
で文字列型のカラムになる -
'AttributeType': 'N'
で数値型のカラムになる
-
2. list_tables() ... テーブル一覧取得
テーブル一覧取得
# region_name には自分のリージョンの名前を書いてね
client = boto3.client('dynamodb', region_name='us-east-2')
client.list_tables()['TableNames']
3. get_item() ... 1件取得
レコード1件取得
table_name = 'hoge_table'
table = dynamodb.Table(table_name)
table.get_item(
Key={'primary_key_name': 1, 'column_name': 'string_value'}
).get('Item')
4. scan() ... 全件取得後にフィルタリング
全件取得してからAttributeでフィルタリングします
query()と比べて余分にリソースを食う(AWS利用料が余分にかかる)と言われている
全件取得後にフィルタリング
table_name = 'hoge_table'
table = dynamodb.Table(table_name)
table.scan(
FilterExpression=Attr('column_name').eq('string_value'),
Limit=1 # 件数を指定できる。別に1件である必要はなし。
).get('Items')
5. query() ... 条件検索
条件に合致するレコードのみ取得します
条件検索
table_name = 'hoge_table'
table = dynamodb.Table(table_name)
# カラム名は適当です
table.query(
KeyConditionExpression=Key('column_name1').eq('string_value') & Key('column_name2').between(from_str, to_str)
)
(なんかうまいとこで改行したいけど、どうしてもLintツールに怒られてしまう...)
6. batch_writer() ... まとめて insert
table_name = 'hoge_table'
table = dynamodb.Table(table_name)
items_source = [
{'primary_key_name': 1, 'column_name': 'hoge1'},
{'primary_key_name': 2, 'column_name': 'hoge2'},
...(略)
{'primary_key_name': 9, 'column_name': 'hoge9'}
]
# `overwrite_by_pkeys` に指定したカラムについては、仮に重複していた場合、レコードを上書きしてくれる
# これを指定しない場合は、値が重複していた時エラーを吐いてくれる
with table.batch_writer(overwrite_by_pkeys=['primary_key_name', 'column_name']) as batch:
for item in items_source:
batch.put_item(Item=item)