boto3でDynamoDBを操作する
DynamoDBのセットアップ
DockerでローカルにDynamoDBを構築しました。 docs.aws.amazon.com
JavaScript Shell
ブラウザでhttp://localhost:(ポート番号)/shell/にアクセスすると、CUIでDynamoDBを操作できます。 設定ダイアログを開いてdocker composeで設定したアクセスキーを入力します。
テーブルの作成
JavaScript Shellがテンプレートを提供しているので、パラメータを変えるだけでテーブルを作れます。
var params = { TableName: 'dummy_table', KeySchema: [ // The type of of schema. Must start with a HASH type, with an optional second RANGE. { // Required HASH type attribute AttributeName: 'key', KeyType: 'HASH', }, ], AttributeDefinitions: [ // The names and types of all primary and index key attributes only { AttributeName: 'key', AttributeType: 'S', // (S | N | B) for string, number, binary }, ], ProvisionedThroughput: { // required provisioned throughput for the table ReadCapacityUnits: 10, WriteCapacityUnits: 1, }, }; dynamodb.createTable(params, function(err, data) { if (err) ppJson(err); // an error occurred else ppJson(data); // successful response });
boto3を使ってDynamoDBを操作
dataclassを使ってCRUD操作を行えるラッパーを作ってみました。
from boto3.session import Session from abc import ABCMeta, abstractmethod from typing import Dict import dataclasses from dataclasses import dataclass @dataclass() class ATable(metaclass=ABCMeta): @property @abstractmethod def table_name(self) -> str: pass @property @abstractmethod def keys(self) -> Dict[str, str]: pass class DynamoDBWrapper: def __init__(self, session: Session, uri): self.resource = session.resource(service_name='dynamodb', endpoint_url=uri) self.client = session.client(service_name='dynamodb', endpoint_url=uri) def insert(self, table: ATable): target = self.resource.Table(table.table_name) return target.put_item(Item=dataclasses.asdict(table)) def select(self, table: ATable): target = self.resource.Table(table.table_name) return target.get_item(Key=table.keys) def update(self, table: ATable): target = self.resource.Table(table.table_name) dict_table = dataclasses.asdict(table) expression, names, values = "", {}, {} for k, v in dict_table.items(): key_name = f'#{k}' value_name = f':{k}' if k in table.keys.keys(): continue if expression == "": expression = f'set {key_name}={value_name}' else: expression += f', {key_name}={value_name}' names[key_name] = k values[value_name] = v return target.update_item(Key=table.keys, UpdateExpression=expression, ExpressionAttributeNames=names, ExpressionAttributeValues=values) def delete(self, table: ATable): target = self.resource.Table(table.table_name) return target.delete_item(Key=table.keys) @dataclass() class DummyTable(ATable): key: str str_value: str dict_value: Dict @property def table_name(self) -> str: return 'dummy_table' @property def keys(self) -> Dict[str, str]: return {'key': self.key} if __name__ == "__main__": sess = Session(aws_access_key_id='DUMMYIDEXAMPLE', aws_secret_access_key='DUMMYEXAMPLEKEY', region_name='us-west-2') ddw = DynamoDBWrapper(sess, 'http://localhost:8000') print(ddw.client.list_tables()) dummy = DummyTable('key1', 'value1', {'value2': 1}) print('INSERT', ddw.insert(dummy)) print('SELECT', ddw.select(dummy)) dummy.str_value = 'value3' dummy.dict_value['value2'] = 2 dummy.dict_value['value4'] = 3 print('UPDATE', ddw.update(dummy)) print('SELECT', ddw.select(dummy)) print('DELETE', ddw.delete(dummy)) print('SCAN', ddw.client.scan(TableName=dummy.table_name))