DynamoDBとの基本的なインタラクションとPythonによるバックアップ・リストア

以下の内容は、Pythonのboto3ライブラリを使用してDynamoDBとやり取りし、テーブルをバックアップおよびリストアする方法について説明します。

まず、boto3をインストールする必要があります。次のコマンドを実行してください：

sudo pip install boto3

boto3はDynamoDBと通信するための多くのメソッドを提供しており、主な操作として次のようなものがあります：

batch_get_item()
batch_write_item()
create_table()
delete_item()
delete_table()
get_item()
put_item()
query()
scan()

以下に、これらの操作をカプセル化したクラスを作成する手順を示します。

1. 初期設定

最初に、AWSの認証情報をJSON形式で保存します。

{
    "region_name": "xxx",
    "aws_access_key_id": "xxx",
    "aws_secret_access_key": "xxx"
}

次に、このJSONファイルを読み込む関数を作成します。

class DynamoDBManager:
    def __init__(self, config_path):
        self.config = self.load_config(config_path)
        self.client = boto3.client(
            'dynamodb',
            region_name=self.config['region_name'],
            aws_access_key_id=self.config['aws_access_key_id'],
            aws_secret_access_key=self.config['aws_secret_access_key']
        )

    def load_config(self, path):
        try:
            with open(path) as file:
                return json.load(file)
        except FileNotFoundError:
            print(f"ERROR: File not found at {path}")
            sys.exit(-1)

2. テーブルの一覧取得

すべてのテーブル名を取得するには、次のコードを使用します。

def list_tables(self):
    tables = []
    last_evaluated = None
    while True:
        if last_evaluated:
            response = self.client.list_tables(ExclusiveStartTableName=last_evaluated)
        else:
            response = self.client.list_tables()
        tables.extend(response['TableNames'])
        last_evaluated = response.get('LastEvaluatedTableName')
        if not last_evaluated:
            break
    return tables

3. テーブル作成

新しいテーブルを作成するには、次のメソッドを使用します。

def create_table(self, table_name, key_schema, attributes, throughput):
    self.client.create_table(
        TableName=table_name,
        KeySchema=key_schema,
        AttributeDefinitions=attributes,
        ProvisionedThroughput=throughput
    )
    waiter = self.client.get_waiter('table_exists')
    waiter.wait(TableName=table_name)

4. データ挿入

データを挿入するには、次の関数を使用します。

def insert_data(self, table_name, item):
    try:
        self.client.put_item(TableName=table_name, Item=item)
    except Exception as e:
        print(f"ERROR: Failed to insert data. Reason: {e}")

5. テーブル削除

テーブルを削除するには、次の関数を使用します。

def delete_table(self, table_name):
    try:
        self.client.delete_table(TableName=table_name)
        print(f"Table {table_name} deleted successfully.")
    except Exception as e:
        print(f"ERROR: Failed to delete table {table_name}. Reason: {e}")

6. テーブルのバックアップとリストア

テーブル構造を保存するには、次の関数を使用します。

def backup_table(self, table_name):
    description = self.client.describe_table(TableName=table_name)['Table']
    backup = {
        "TableName": description["TableName"],
        "KeySchema": description["KeySchema"],
        "AttributeDefinitions": description["AttributeDefinitions"]
    }
    return json.dumps(backup)

同じ構造のテーブルをリストアするには、次の関数を使用します。

def restore_table(self, backup_json):
    backup = json.loads(backup_json)
    self.create_table(
        backup["TableName"],
        backup["KeySchema"],
        backup["AttributeDefinitions"],
        {"ReadCapacityUnits": 5, "WriteCapacityUnits": 5}
    )

以上が、DynamoDBとの基本的なインタラクションとバックアップ・リストアの方法です。

タグ: boto3 dynamodb Python

7月9日 16:53 投稿

異端開発室