概要

なるべく行数や文字数を最小限に抑えつつ、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)

参考にした記事