DynamoDB上のアイテムをコードで管理するPythonスクリプトを作ってみた
こんにちは、CX事業本部の若槻です。
DynamoDBにシステムの定義に必要なアイテム(APIの認可情報など)を登録する場合は、そのアイテムをコード化してGitHubなどでバージョン管理したい場合があります。
しかし、AWS CloudFormationではDynamoDBのテーブル自体をリソースとして管理することはできますが、テーブル上のアイテムについては管理することができません。そこで別途アイテムをコードで管理する方法としては以下が考えられます。
- Terraform(Resource: aws_dynamodb_table_item)
- スクリプトを自作して使う
今回は、後者の方法をとり、DynamoDB上のアイテムをコードで管理するPythonスクリプトを作ってみました。
スクリプト
ディレクトリ構成は以下のようになります。initialize.py
とdeploy.py
の2つのスクリプトを使い、src_data
ディレクトリ配下のJSONファイルでテーブルごとのアイテムを管理します。
. ├── deploy.py ├── initialize.py └── src_data └── <table_name>.json
スクリプトはGitHubにもアップロードしてあります。
ここではそれぞれのスクリプトについて説明をしていきます。
initialize.py
:ソースデータを初期化するスクリプト
スクリプトinitialize.py
の利用は必須ではありません。アイテムを管理したいDynamoDBテーブル上に既にアイテムが登録されている場合に、initialize.py
を実行すればそのテーブル上のアイテムをスキャンして<table_name>.json
ファイルを作成(初期化)することができます。
import os import json import sys from typing import List from decimal import Decimal import boto3 from boto3.resources.base import ServiceResource from boto3.dynamodb.types import Binary def main(table_name: str) -> None: table_items = scan_table(table_name, dynamodb_resource) put_to_src_data(table_name, table_items) def scan_table( table_name: str, dynamodb_resource: ServiceResource ) -> List[dict]: table = dynamodb_resource.Table(table_name) resp = table.scan() table_items = resp['Items'] while 'LastEvaluatedKey' in resp: resp = table.scan( ExclusiveStartKey=resp['LastEvaluatedKey'] ) table_items.extend(resp['Items']) return table_items def default(obj) -> object: if isinstance(obj, Decimal): if int(obj) == obj: return int(obj) else: return float(obj) elif isinstance(obj, Binary): return obj.value elif isinstance(obj, bytes): return obj.decode() elif isinstance(obj, set): return list(obj) try: return str(obj) except Exception: return None def put_to_src_data( table_name: str, table_items: List[dict] ) -> None: f = open(f'src_data/{table_name}.json', 'w') json.dump(table_items, f, default=default, ensure_ascii=False, indent=4) if __name__ == '__main__': dynamodb_resource = boto3.resource('dynamodb') _, table_name = sys.argv main(table_name)
スクリプトは第1引数にテーブル名を指定して実行します。
$ python initialize.py <テーブル名>
動作確認
テーブル名sample-table
、パーティションキーtype
、ソートキーname
のDynamoDBテーブルに以下のようにアイテムが既に登録されているとします。
sample-table
テーブルに対してスクリプトを実行します。
$ python initialize.py sample-table
src_data
ディレクトリ配下にsample-table.json
ファイルが作成されました。テーブル上のアイテムが配列としてファイルに書き込まれています。
$ cat ./src_data/sample-table.json [ { "9th_rank": 3, "name": "ichinose", "type": "cute" }, { "9th_rank": 1, "name": "hojo", "type": "cool" }, { "9th_rank": 2, "name": "sagisawa", "type": "cool" } ]
この<table_name>.json
ファイルが、テーブル上のアイテムをコードで管理するためのソースデータとなります。
deploy.py
:ソースデータをもとにデプロイを行うスクリプト
ソースデータである<table_name>.json
ファイルを編集してdeploy.py
スクリプトを実行することにより、テーブルにアイテムをデプロイ(追加・更新・削除)することができます。
import json import sys from typing import List import boto3 from boto3.resources.base import ServiceResource from botocore.client import BaseClient KEY_SPLITTER = ':' def main( table_name: str, dynamodb_resource: ServiceResource, dynamodb_client: BaseClient ) -> None: partition_key, sort_key = get_table_keys(table_name, dynamodb_client) src_items = get_src_items(table_name) put_to_table(table_name, src_items, dynamodb_resource) table_items = scan_table(table_name, dynamodb_resource) diff_keys = get_diff_keys(src_items, table_items, partition_key, sort_key) delete_diff_items( table_name, partition_key, sort_key, diff_keys, dynamodb_resource ) def get_table_keys(table_name: str, dynamodb_client: BaseClient) -> (str, str): resp = dynamodb_client.describe_table(TableName=table_name) sort_key = '' for key_schema in resp['Table']['KeySchema']: if key_schema['KeyType'] == 'HASH': partition_key = key_schema['AttributeName'] elif key_schema['KeyType'] == 'RANGE': sort_key = key_schema['AttributeName'] return partition_key, sort_key def get_src_items( table_name: str ) -> List[dict]: return json.load( open(f'./src_data/{table_name}.json') ) def put_to_table( table_name: str, src_items: List[dict], dynamodb_resource: ServiceResource ) -> None: table = dynamodb_resource.Table(table_name) with table.batch_writer() as batch: for item in src_items: batch.put_item( Item=item ) def scan_table( table_name: str, dynamodb_resource: ServiceResource ) -> List[dict]: table = dynamodb_resource.Table(table_name) resp = table.scan(Limit=10) table_items = resp['Items'] while 'LastEvaluatedKey' in resp: resp = table.scan( ExclusiveStartKey=resp['LastEvaluatedKey'], Limit=10 ) table_items += resp['Items'] return table_items def get_diff_keys( src_items: List[dict], table_items: List[dict], partition_key: str, sort_key: str ) -> List[str]: src_keys = [] for src_item in src_items: src_keys.append( src_item[partition_key] + KEY_SPLITTER + (src_item[sort_key] if sort_key != '' else '') ) table_keys = [] for table_item in table_items: table_keys.append( table_item[partition_key] + KEY_SPLITTER + (table_item[sort_key] if sort_key != '' else '') ) return list(set(table_keys) - set(src_keys)) def delete_diff_items( table_name: str, partition_key: str, sort_key: str, diff_keys: List[str], dynamodb_resource: ServiceResource ) -> None: table = dynamodb_resource.Table(table_name) for diff_key in diff_keys: key = {partition_key: diff_key.split(KEY_SPLITTER)[0]} if sort_key != '': sort_key_d = {sort_key: diff_key.split(KEY_SPLITTER)[1]} key.update(**sort_key_d) table.delete_item( Key=key ) if __name__ == '__main__': dynamodb_resource = boto3.resource('dynamodb') dynamodb_client = boto3.client('dynamodb') _, table_name = sys.argv main(table_name, dynamodb_resource, dynamodb_client)
スクリプトは第1引数にテーブル名を指定して実行します。
$ python deploy.py <テーブル名>
動作確認
- アイテムの追加
先ほど作成したsrc_data/sample-table.json
を以下のように更新します。アイテムを1つ追加しています。
[ { "9th_rank": 3, "name": "ichinose", "type": "cute" }, { "9th_rank": 1, "name": "hojo", "type": "cool" }, { "9th_rank": 2, "name": "sagisawa", "type": "cool" }, { "9th_rank": 4, "name": "kamiya", "type": "cool" } ]
sample-table
テーブルに対してスクリプトを実行します。
$ python deploy.py sample-table
コンソール画面を見ると、ソースデータに追加したアイテムがテーブルに追加できています。
- アイテムの編集
src_data/sample-table.json
を以下のように更新します。各アイテムに8th_rank
属性を追加しています。
[ { "9th_rank": 3, "name": "ichinose", "type": "cute", "8th_rank": 6 }, { "9th_rank": 1, "name": "hojo", "type": "cool", "8th_rank": 2 }, { "9th_rank": 2, "name": "sagisawa", "type": "cool", "8th_rank": 7 }, { "9th_rank": 4, "name": "kamiya", "type": "cool", "8th_rank": 16 } ]
sample-table
テーブルに対してスクリプトを実行します。
$ python deploy.py sample-table
コンソール画面を見ると、ソースデータでのアイテムの更新がテーブル上でも反映されています。
- アイテムの削除
src_data/sample-table.json
を以下のように更新します。アイテムを1つ削除しています。
[ { "9th_rank": 1, "name": "hojo", "type": "cool", "8th_rank": 2 }, { "9th_rank": 2, "name": "sagisawa", "type": "cool", "8th_rank": 7 }, { "9th_rank": 4, "name": "kamiya", "type": "cool", "8th_rank": 16 } ]
sample-table
テーブルに対してスクリプトを実行します。
$ python deploy.py sample-table
コンソール画面を見ると、ソースデータで~~失踪~~削除したアイテムがテーブル上でも削除できています。
※ここまで、ソートキーがあるテーブルで動作確認をしましたが、ソートキーなしのテーブルでも動作することは確認済みです。
おわりに
以上、DynamoDB上のアイテムをコードで管理するPythonスクリプトのご紹介でした。あとはsrc_data
配下のJSONファイルをGitHubで管理するなり、スクリプトをCI/CDサービス上で動かしてデプロイ自動化するなりすれば捗るかと思います。
参考
以上