yoraba build

備忘録を兼ねた技術ブログ

boto3でDynamoDBを操作する

DynamoDBのセットアップ

DockerでローカルにDynamoDBを構築しました。 docs.aws.amazon.com

f:id:yoraba:20200704022222p:plain

JavaScript Shell

ブラウザでhttp://localhost:(ポート番号)/shell/にアクセスすると、CUIでDynamoDBを操作できます。 f:id:yoraba:20200704022730p:plain 設定ダイアログを開いてdocker composeで設定したアクセスキーを入力します。

テーブルの作成

JavaScript Shellがテンプレートを提供しているので、パラメータを変えるだけでテーブルを作れます。

f:id:yoraba:20200704023938p:plain

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))