Example script to import DynamoDB table data

import_dynamodb_data_script.py

import json
import boto3

# Specify your existing (export) and new (import) table names and the region your tables are in
export_table_name = "ExportDataTable"
import_table_name = "ImportDataTable"
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
table = dynamodb.Table(import_table_name)

def import_table_data():
    table_data_export = read_data_from_export_file()
    # You can add functions here if you need to do any additional data transformations
    # For example, for my migration I needed to generate Cognito IDs for each user
    # table_data_export = generate_cognito_ids(table_data_export)
    write_data_to_dynamodb(table_data_export)

def read_data_from_export_file():
    # Read data from the DynamodB export file you created with the export script
    table_data_export = []
    with open(f"data_export_{export_table_name}.json", "r") as f:
        contents = f.read()
        table_data_export = json.loads(contents)
        
    return table_data_export

def write_data_to_dynamodb(table_data_export): 
    failed_users_list = []
    succeeded_users_count = 0

    for user in table_data_export:
        try:
            # For each item in the export table, put a new item in the import table using the new data model structure
            # Use a ConditionExpression to only put the item if a user with the same cognito_id does not already exist
            response = table.put_item(
                Item={
                        'PK': "USER#" + user['cognito_id'],
                        'SK': "USER#" + user['cognito_id'],
                        'emailAddress': user['SubscriberEmail'],
                        'dateCreated': user['DateSubscribed'],
                        'lastLogin': "",
                        'userAlias': "Not set",
                        'userAliasPinyin': "Not set",
                        'userAliasEmoji': "Not set",
                        'characterSetPreference': user['CharacterSet'],
                        'GSI1PK': "USER",
                        'GSI1SK': "USER#" + user['cognito_id']
                },
                ConditionExpression='attribute_not_exists(PK)'
            )
            print(f"Create contact in DynamoDB {user['cognito_id']}. Response: {response['ResponseMetadata']['HTTPStatusCode']}")
            succeeded_users_count += 1

        except Exception as e:
            print(f"Error: Failed to create contact in DynamoDB, {user['cognito_id']}. Error: {e}")
            failed_users_list.append(user)

    if failed_users_list:
        print('Failed users: ', failed_users_list)
    print('Succeeded users count: ', succeeded_users_count)
    print('Import completed!')

    return

# Example of an additional migration step
def generate_cognito_ids(table_data_export):
    # Create Cognito profile for each user and append Cognito IDs as cognito_id to user data 
    return table_data_export

import_table_data()