(
    dynamo,
    sleep_interval,
    source_table,
    destination_table,
    write_capacity,
    billing_mode,
)
Restore table
Restore table
Restore table
def do_restore(
    dynamo,
    sleep_interval,
    source_table,
    destination_table,
    write_capacity,
    billing_mode,
):
    """
    Restore table
    """
    logging.info(
        "Starting restore for " + source_table + " to " + destination_table + ".."
    )

    # create table using schema
    # restore source_table from dump directory if it exists else try current working directory
    if os.path.exists("%s/%s" % (args.dumpPath, source_table)):
        dump_data_path = args.dumpPath
    else:
        logging.info(
            'Cannot find "./%s/%s", Now trying current working directory..'
            % (args.dumpPath, source_table)
        )
        if os.path.exists("%s/%s" % (CURRENT_WORKING_DIR, source_table)):
            dump_data_path = CURRENT_WORKING_DIR
        else:
            logging.info(
                'Cannot find "%s/%s" directory containing dump files!'
                % (CURRENT_WORKING_DIR, source_table)
            )
            sys.exit(1)
    table_data = json.load(
        open(dump_data_path + os.sep + source_table + os.sep + SCHEMA_FILE)
    )
    table = table_data["Table"]
    table_attribute_definitions = table["AttributeDefinitions"]
    table_table_name = destination_table
    table_key_schema = table["KeySchema"]
    original_read_capacity = table["ProvisionedThroughput"]["ReadCapacityUnits"]
    original_write_capacity = table["ProvisionedThroughput"]["WriteCapacityUnits"]
    table_local_secondary_indexes = table.get("LocalSecondaryIndexes")
    table_global_secondary_indexes = table.get("GlobalSecondaryIndexes")

    # override table write capacity if specified, else use RESTORE_WRITE_CAPACITY if original
    # write capacity is lower
    if write_capacity is None:
        if original_write_capacity < RESTORE_WRITE_CAPACITY:
            write_capacity = RESTORE_WRITE_CAPACITY
        else:
            write_capacity = original_write_capacity

    if original_write_capacity == 0:
        original_write_capacity = RESTORE_WRITE_CAPACITY

    # ensure that read capacity is at least RESTORE_READ_CAPACITY
    if original_read_capacity < RESTORE_READ_CAPACITY:
        read_capacity = RESTORE_WRITE_CAPACITY
    else:
        read_capacity = original_read_capacity

    if original_read_capacity == 0:
        original_read_capacity = RESTORE_READ_CAPACITY

    # override GSI write capacities if specified, else use RESTORE_WRITE_CAPACITY if original
    # write capacity is lower
    original_gsi_write_capacities = []
    original_gsi_read_capacities = []
    if table_global_secondary_indexes is not None:
        for gsi in table_global_secondary_indexes:
            # keeps track of original gsi write capacity units. If provisioned capacity is 0, set to
            # RESTORE_WRITE_CAPACITY as fallback given that 0 is not allowed for write capacities
            original_gsi_write_capacity = gsi["ProvisionedThroughput"][
                "WriteCapacityUnits"
            ]
            if original_gsi_write_capacity == 0:
                original_gsi_write_capacity = RESTORE_WRITE_CAPACITY

            original_gsi_write_capacities.append(original_gsi_write_capacity)

            if gsi["ProvisionedThroughput"]["WriteCapacityUnits"] < int(write_capacity):
                gsi["ProvisionedThroughput"]["WriteCapacityUnits"] = int(write_capacity)

            # keeps track of original gsi read capacity units. If provisioned capacity is 0, set to
            # RESTORE_READ_CAPACITY as fallback given that 0 is not allowed for read capacities
            original_gsi_read_capacity = gsi["ProvisionedThroughput"][
                "ReadCapacityUnits"
            ]
            if original_gsi_read_capacity == 0:
                original_gsi_read_capacity = RESTORE_READ_CAPACITY

            original_gsi_read_capacities.append(original_gsi_read_capacity)

            if (
                gsi["ProvisionedThroughput"]["ReadCapacityUnits"]
                < RESTORE_READ_CAPACITY
            ):
                gsi["ProvisionedThroughput"][
                    "ReadCapacityUnits"
                ] = RESTORE_READ_CAPACITY

    # temp provisioned throughput for restore
    table_provisioned_throughput = {
        "ReadCapacityUnits": int(read_capacity),
        "WriteCapacityUnits": int(write_capacity),
    }

    optional_args = {}
    if billing_mode == PROVISIONED_BILLING_MODE:
        optional_args["ProvisionedThroughput"] = table_provisioned_throughput

    if not args.dataOnly:

        logging.info(
            "Creating "
            + destination_table
            + " table with temp write capacity of "
            + str(write_capacity)
        )

        if table_local_secondary_indexes is not None:
            optional_args["LocalSecondaryIndexes"] = table_local_secondary_indexes

        if table_global_secondary_indexes is not None:
            optional_args["GlobalSecondaryIndexes"] = [
                prepare_gsi_for_restore(gsi) for gsi in table_global_secondary_indexes
            ]

        while True:
            try:
                dynamo.create_table(
                    AttributeDefinitions=table_attribute_definitions,
                    TableName=table_table_name,
                    KeySchema=table_key_schema,
                    BillingMode=billing_mode,
                    **optional_args
                )
                break
            except dynamo.exceptions.LimitExceededException:
                logging.info(
                    "Limit exceeded, retrying creation of " + destination_table + ".."
                )
                time.sleep(sleep_interval)
            except dynamo.exceptions.ProvisionedThroughputExceededException:
                logging.info(
                    "Control plane limit exceeded, "
                    "retrying creation of " + destination_table + ".."
                )
                time.sleep(sleep_interval)
            except dynamo.exceptions.ClientError as e:
                logging.exception(e)
                sys.exit(1)

        # wait for table creation completion
        wait_for_active_table(dynamo, destination_table, "created")
    elif not args.skipThroughputUpdate:
        # update provisioned capacity
        if int(write_capacity) > original_write_capacity:
            update_provisioned_throughput(
                dynamo, destination_table, original_read_capacity, write_capacity, False
            )

    if not args.schemaOnly:
        # read data files
        logging.info("Restoring data for " + destination_table + " table..")
        data_file_list = os.listdir(
            dump_data_path + os.sep + source_table + os.sep + DATA_DIR + os.sep
        )
        data_file_list.sort()

        for data_file in data_file_list:
            logging.info("Processing " + data_file + " of " + destination_table)
            items = []
            item_data = json.load(
                open(
                    dump_data_path
                    + os.sep
                    + source_table
                    + os.sep
                    + DATA_DIR
                    + os.sep
                    + data_file
                )
            )
            items.extend(item_data["Items"])

            # batch write data
            put_requests = []
            while len(items) > 0:
                put_requests.append({"PutRequest": {"Item": items.pop(0)}})

                # flush every MAX_BATCH_WRITE
                if len(put_requests) == MAX_BATCH_WRITE:
                    logging.debug(
                        "Writing next "
                        + str(MAX_BATCH_WRITE)
                        + " items to "
                        + destination_table
                        + ".."
                    )
                    batch_write(
                        dynamo,
                        BATCH_WRITE_SLEEP_INTERVAL,
                        destination_table,
                        put_requests,
                    )
                    del put_requests[:]

            # flush remainder
            if len(put_requests) > 0:
                batch_write(
                    dynamo, BATCH_WRITE_SLEEP_INTERVAL, destination_table, put_requests
                )

        if not args.skipThroughputUpdate:
            # revert to original table write capacity if it has been modified
            if (
                int(write_capacity) != original_write_capacity
                or int(read_capacity) != original_read_capacity
            ):
                update_provisioned_throughput(
                    dynamo,
                    destination_table,
                    original_read_capacity,
                    original_write_capacity,
                    False,
                )

            # loop through each GSI to check if it has changed and update if necessary
            if table_global_secondary_indexes is not None:
                gsi_data = []
                for gsi in table_global_secondary_indexes:
                    wcu = gsi["ProvisionedThroughput"]["WriteCapacityUnits"]
                    rcu = gsi["ProvisionedThroughput"]["ReadCapacityUnits"]
                    original_gsi_write_capacity = original_gsi_write_capacities.pop(0)
                    original_gsi_read_capacity = original_gsi_read_capacities.pop(0)
                    if (
                        original_gsi_write_capacity != wcu
                        or original_gsi_read_capacity != rcu
                    ):
                        gsi_data.append(
                            {
                                "Update": {
                                    "IndexName": gsi["IndexName"],
                                    "ProvisionedThroughput": {
                                        "ReadCapacityUnits": int(
                                            original_gsi_read_capacity
                                        ),
                                        "WriteCapacityUnits": int(
                                            original_gsi_write_capacity
                                        ),
                                    },
                                }
                            }
                        )

                if gsi_data:
                    logging.info(
                        "Updating "
                        + destination_table
                        + " global secondary indexes write and read capacities as necessary.."
                    )
                    while True:
                        try:
                            dynamo.update_table(
                                TableName=destination_table,
                                GlobalSecondaryIndexUpdates=gsi_data,
                            )
                            break
                        except dynamo.exceptions.LimitExceededException:
                            logging.info(
                                "Limit exceeded, retrying updating throughput of"
                                "GlobalSecondaryIndexes in " + destination_table + ".."
                            )
                            time.sleep(sleep_interval)
                        except dynamo.exceptions.ProvisionedThroughputExceededException:
                            logging.info(
                                "Control plane limit exceeded, retrying updating throughput of"
                                "GlobalSecondaryIndexes in " + destination_table + ".."
                            )
                            time.sleep(sleep_interval)

        # wait for table to become active
        wait_for_active_table(dynamo, destination_table, "active")

        logging.info(
            "Restore for "
            + source_table
            + " to "
            + destination_table
            + " table completed. Time taken: "
            + str(datetime.datetime.now().replace(microsecond=0) - start_time)
        )
    else:
        logging.info(
            "Empty schema of "
            + source_table
            + " table created. Time taken: "
            + str(datetime.datetime.now().replace(microsecond=0) - start_time)
        )
def do_restore ( dynamo , sleep_interval , source_table , destination_table , write_capacity , billing_mode , ) : logging . info ( "Starting restore for " + source_table + " to " + destination_table + ".." ) # create table using schema # restore source_table from dump directory if it exists else try current working directory if os . path . exists ( "%s/%s" % ( args . dumpPath , source_table ) ) : dump_data_path = args . dumpPath else : logging . info ( 'Cannot find "./%s/%s", Now trying current working directory..' % ( args . dumpPath , source_table ) ) if os . path . exists ( "%s/%s" % ( CURRENT_WORKING_DIR , source_table ) ) : dump_data_path = CURRENT_WORKING_DIR else : logging . info ( 'Cannot find "%s/%s" directory containing dump files!' % ( CURRENT_WORKING_DIR , source_table ) ) sys . exit ( 1 ) table_data = json . load ( open ( dump_data_path + os . sep + source_table + os . sep + SCHEMA_FILE ) ) table = table_data [ "Table" ] table_attribute_definitions = table [ "AttributeDefinitions" ] table_table_name = destination_table table_key_schema = table [ "KeySchema" ] original_read_capacity = table [ "ProvisionedThroughput" ] [ "ReadCapacityUnits" ] original_write_capacity = table [ "ProvisionedThroughput" ] [ "WriteCapacityUnits" ] table_local_secondary_indexes = table . get ( "LocalSecondaryIndexes" ) table_global_secondary_indexes = table . get ( "GlobalSecondaryIndexes" ) # override table write capacity if specified, else use RESTORE_WRITE_CAPACITY if original # write capacity is lower if write_capacity is None : if original_write_capacity < RESTORE_WRITE_CAPACITY : write_capacity = RESTORE_WRITE_CAPACITY else : write_capacity = original_write_capacity if original_write_capacity == 0 : original_write_capacity = RESTORE_WRITE_CAPACITY # ensure that read capacity is at least RESTORE_READ_CAPACITY if original_read_capacity < RESTORE_READ_CAPACITY : read_capacity = RESTORE_WRITE_CAPACITY else : read_capacity = original_read_capacity if original_read_capacity == 0 : original_read_capacity = RESTORE_READ_CAPACITY # override GSI write capacities if specified, else use RESTORE_WRITE_CAPACITY if original # write capacity is lower original_gsi_write_capacities = [ ] original_gsi_read_capacities = [ ] if table_global_secondary_indexes is not None : for gsi in table_global_secondary_indexes : # keeps track of original gsi write capacity units. If provisioned capacity is 0, set to # RESTORE_WRITE_CAPACITY as fallback given that 0 is not allowed for write capacities original_gsi_write_capacity = gsi [ "ProvisionedThroughput" ] [ "WriteCapacityUnits" ] if original_gsi_write_capacity == 0 : original_gsi_write_capacity = RESTORE_WRITE_CAPACITY original_gsi_write_capacities . append ( original_gsi_write_capacity ) if gsi [ "ProvisionedThroughput" ] [ "WriteCapacityUnits" ] < int ( write_capacity ) : gsi [ "ProvisionedThroughput" ] [ "WriteCapacityUnits" ] = int ( write_capacity ) # keeps track of original gsi read capacity units. If provisioned capacity is 0, set to # RESTORE_READ_CAPACITY as fallback given that 0 is not allowed for read capacities original_gsi_read_capacity = gsi [ "ProvisionedThroughput" ] [ "ReadCapacityUnits" ] if original_gsi_read_capacity == 0 : original_gsi_read_capacity = RESTORE_READ_CAPACITY original_gsi_read_capacities . append ( original_gsi_read_capacity ) if ( gsi [ "ProvisionedThroughput" ] [ "ReadCapacityUnits" ] < RESTORE_READ_CAPACITY ) : gsi [ "ProvisionedThroughput" ] [ "ReadCapacityUnits" ] = RESTORE_READ_CAPACITY # temp provisioned throughput for restore table_provisioned_throughput = { "ReadCapacityUnits" : int ( read_capacity ) , "WriteCapacityUnits" : int ( write_capacity ) , } optional_args = { } if billing_mode == PROVISIONED_BILLING_MODE : optional_args [ "ProvisionedThroughput" ] = table_provisioned_throughput if not args . dataOnly : logging . info ( "Creating " + destination_table + " table with temp write capacity of " + str ( write_capacity ) ) if table_local_secondary_indexes is not None : optional_args [ "LocalSecondaryIndexes" ] = table_local_secondary_indexes if table_global_secondary_indexes is not None : optional_args [ "GlobalSecondaryIndexes" ] = [ prepare_gsi_for_restore ( gsi ) for gsi in table_global_secondary_indexes ] while True : try : dynamo . create_table ( AttributeDefinitions = table_attribute_definitions , TableName = table_table_name , KeySchema = table_key_schema , BillingMode = billing_mode , * * optional_args ) break except dynamo . exceptions . LimitExceededException : logging . info ( "Limit exceeded, retrying creation of " + destination_table + ".." ) time . sleep ( sleep_interval ) except dynamo . exceptions . ProvisionedThroughputExceededException : logging . info ( "Control plane limit exceeded, " "retrying creation of " + destination_table + ".." ) time . sleep ( sleep_interval ) except dynamo . exceptions . ClientError as e : logging . exception ( e ) sys . exit ( 1 ) # wait for table creation completion wait_for_active_table ( dynamo , destination_table , "created" ) elif not args . skipThroughputUpdate : # update provisioned capacity if int ( write_capacity ) > original_write_capacity : update_provisioned_throughput ( dynamo , destination_table , original_read_capacity , write_capacity , False ) if not args . schemaOnly : # read data files logging . info ( "Restoring data for " + destination_table + " table.." ) data_file_list = os . listdir ( dump_data_path + os . sep + source_table + os . sep + DATA_DIR + os . sep ) data_file_list . sort ( ) for data_file in data_file_list : logging . info ( "Processing " + data_file + " of " + destination_table ) items = [ ] item_data = json . load ( open ( dump_data_path + os . sep + source_table + os . sep + DATA_DIR + os . sep + data_file ) ) items . extend ( item_data [ "Items" ] ) # batch write data put_requests = [ ] while len ( items ) > 0 : put_requests . append ( { "PutRequest" : { "Item" : items . pop ( 0 ) } } ) # flush every MAX_BATCH_WRITE if len ( put_requests ) == MAX_BATCH_WRITE : logging . debug ( "Writing next " + str ( MAX_BATCH_WRITE ) + " items to " + destination_table + ".." ) batch_write ( dynamo , BATCH_WRITE_SLEEP_INTERVAL , destination_table , put_requests , ) del put_requests [ : ] # flush remainder if len ( put_requests ) > 0 : batch_write ( dynamo , BATCH_WRITE_SLEEP_INTERVAL , destination_table , put_requests ) if not args . skipThroughputUpdate : # revert to original table write capacity if it has been modified if ( int ( write_capacity ) != original_write_capacity or int ( read_capacity ) != original_read_capacity ) : update_provisioned_throughput ( dynamo , destination_table , original_read_capacity , original_write_capacity , False , ) # loop through each GSI to check if it has changed and update if necessary if table_global_secondary_indexes is not None : gsi_data = [ ] for gsi in table_global_secondary_indexes : wcu = gsi [ "ProvisionedThroughput" ] [ "WriteCapacityUnits" ] rcu = gsi [ "ProvisionedThroughput" ] [ "ReadCapacityUnits" ] original_gsi_write_capacity = original_gsi_write_capacities . pop ( 0 ) original_gsi_read_capacity = original_gsi_read_capacities . pop ( 0 ) if ( original_gsi_write_capacity != wcu or original_gsi_read_capacity != rcu ) : gsi_data . append ( { "Update" : { "IndexName" : gsi [ "IndexName" ] , "ProvisionedThroughput" : { "ReadCapacityUnits" : int ( original_gsi_read_capacity ) , "WriteCapacityUnits" : int ( original_gsi_write_capacity ) , } , } } ) if gsi_data : logging . info ( "Updating " + destination_table + " global secondary indexes write and read capacities as necessary.." ) while True : try : dynamo . update_table ( TableName = destination_table , GlobalSecondaryIndexUpdates = gsi_data , ) break except dynamo . exceptions . LimitExceededException : logging . info ( "Limit exceeded, retrying updating throughput of" "GlobalSecondaryIndexes in " + destination_table + ".." ) time . sleep ( sleep_interval ) except dynamo . exceptions . ProvisionedThroughputExceededException : logging . info ( "Control plane limit exceeded, retrying updating throughput of" "GlobalSecondaryIndexes in " + destination_table + ".." ) time . sleep ( sleep_interval ) # wait for table to become active wait_for_active_table ( dynamo , destination_table , "active" ) logging . info ( "Restore for " + source_table + " to " + destination_table + " table completed. Time taken: " + str ( datetime . datetime . now ( ) . replace ( microsecond = 0 ) - start_time ) ) else : logging . info ( "Empty schema of " + source_table + " table created. Time taken: " + str ( datetime . datetime . now ( ) . replace ( microsecond = 0 ) - start_time ) )