Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use restore_table_to_point_in_time method in localstack

Best Python code snippet using localstack_python

restore_dynamodb.py

Source:restore_dynamodb.py

...36        describe_backup_json = json.loads(describe_backup_output)37        actual_backup_status = describe_backup_json["BackupDescription"]["BackupDetails"]["BackupStatus"]38        logging.info("%s backup status is %s", backup_name, actual_backup_status)39    logging.info("OnDemand Backup %s is %s now", backup_name, actual_backup_status)40def restore_table_to_point_in_time(source_table_name, target_table_name, restoration_point):41    logging.info("Starting point in time recovery of table %s to %s", source_table_name, target_table_name)42    output = os.popen("aws dynamodb restore-table-to-point-in-time --source-table-name " + source_table_name + " --target-table-name " + target_table_name + " --restore-date-time " + restoration_point).read()43    logging.info(output)44    # It will poll every 20 seconds until a successful state has been reached. This will exit with a return code of 255 after 25 failed checks.45    table_exist_response = os.popen("aws dynamodb wait table-exists --table-name " + target_table_name).read()46    logging.info(table_exist_response)47def wait_for_table_to_be_in_active_status(table_name):48    describe_table_output = os.popen("aws dynamodb describe-table --table-name " + table_name).read()49    describe_table_output_json = json.loads(describe_table_output)50    actual_table_status = describe_table_output_json["Table"]["TableStatus"]51    expected_table_status = "ACTIVE"52    while expected_table_status != actual_table_status:53        describe_table_output = os.popen("aws dynamodb describe-table --table-name " + table_name).read()54        describe_table_output_json = json.loads(describe_table_output)55        actual_table_status = describe_table_output_json["Table"]["TableStatus"]56        logging.info("%s table status is %s", table_name, actual_table_status)57        time.sleep(5) #5 sec58def delete_table_if_exist(table_name, aws_profile):59    session = boto3.Session(profile_name=aws_profile)60    dynamodb_client = session.client('dynamodb')61    try:62        dynamodb_client.delete_table(TableName=table_name)63        os.system("aws dynamodb wait table-not-exists --table-name " + table_name)64        logging.info("Table %s deleted", table_name)65    except dynamodb_client.exceptions.ResourceNotFoundException:66        logging.info("Table %s doesn't exist", table_name)67def delete_table_using_aws_cli(table_name):68    delete_table_output = os.popen("aws dynamodb delete-table --table-name " + table_name).read()69    logging.info(delete_table_output)70    os.system("aws dynamodb wait table-not-exists --table-name " + table_name)71    logging.info("%s table successfully deleted!" %table_name)72def restore_table_from_backup(target_table_name, backup_arn):73    restore_table_output = os.popen("aws dynamodb restore-table-from-backup --target-table-name " +target_table_name+ " --backup-arn " + backup_arn).read()74    logging.info(restore_table_output)75    os.system("aws dynamodb wait table-exists --table-name " + target_table_name)76def describe_table(table_name):77    os.popen("aws dynamodb describe-table --table-name " + table_name).read()78def is_table_exist(table_name, aws_profile):79    session = boto3.Session(profile_name=aws_profile)80    dynamodb_client = session.client('dynamodb')81    is_table_exist = True82    try:83        dynamodb_client.describe_table(TableName=table_name)84    except dynamodb_client.exceptions.ResourceNotFoundException:85        is_table_exist = False86    if is_table_exist == False:87        logging.info("Source table %s does not exist!", table_name)88        raise Exception("Source table doesn't exist")89def delete_record_by_lro_name(table_name, name):90    try:91        client = boto3.client('dynamodb')92        response = client.delete_item(TableName=table_name, Key={"name": { 'S': name}})93        logging.info("name: %s successfully deleted. Response: %s", name, response)94    except client.ClientError as err:95        logging.info(err)96        sys.exit()97    else:98        return response99def delete_records_with_last_updated_time_after_recovery_point(table_name, recovery_point, aws_profile):100    try:101        session = boto3.Session(profile_name=aws_profile)102        client = session.client('dynamodb')103        dynamopaginator = client.get_paginator('scan')104        dynamoresponse = dynamopaginator.paginate(105            TableName = table_name,106            Select='ALL_ATTRIBUTES',107            ReturnConsumedCapacity='NONE',108            ConsistentRead=True109        )110        for page in dynamoresponse:111            for item in page['Items']:112                if item.get('lastUpdatedTime') == None:113                    continue114                logging.info(item)115                name = item.get('name').get('S')116                last_updated_time_str = item.get('lastUpdatedTime').get('S')117                last_updated_time = datetime.strptime(last_updated_time_str, "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=timezone.utc)118                if recovery_point.replace(tzinfo=timezone.utc) < last_updated_time.replace(tzinfo=timezone.utc):119                    logging.info("Last Updated Time %s greater than recovery point %s", last_updated_time, recovery_point)120                    delete_record_by_lro_name(table_name, name)121    except Exception:122        logging.info("Error while deleting records")123        raise Exception("Error while deleting records!")124def log_arguments(source_client_name, target_client_name, aws_profile, restore_datetime_epoch, env):125    logging.info("Restore datetime epoch: %s", restore_datetime_epoch)126    logging.info("Source client Name: %s", source_client_name)127    logging.info("Target client Name: %s", target_client_name)128    logging.info("Aws profile: %s", aws_profile)129    logging.info("Source table name: %s", utils.get_lro_store_table_name(source_client_name, env))130    logging.info("Target table name: %s", utils.get_lro_store_table_name(target_client_name, env))131def enable_point_in_time_recovery_on_table_using_aws_cli(table_name):132    output = os.popen("aws dynamodb update-continuous-backups --table-name " + table_name + " --point-in-time-recovery-specification PointInTimeRecoveryEnabled=true").read()133    logging.info("PITR enabled on table %s, Response: %s", table_name, output)134def enable_point_in_time_recovery_on_table(table_name, aws_profile):135    session = boto3.Session(profile_name=aws_profile)136    client = session.client('dynamodb')137    response = client.update_continuous_backups(138        TableName=table_name,139        PointInTimeRecoverySpecification={140            'PointInTimeRecoveryEnabled': True141        }142    )143    logging.info("PITR enabled on table %s, Response: %s", table_name, response)144def table_arn(table_name, aws_profile):145    session = boto3.Session(profile_name=aws_profile)146    resource = session.resource('dynamodb')147    return resource.Table(table_name).table_arn148def get_pitr_status(table_name, aws_profile):149    session = boto3.Session(profile_name=aws_profile)150    dynamodb_client = session.client('dynamodb')151    continuous_backup_json_response = json.loads(json.dumps(dynamodb_client.describe_continuous_backups(TableName=table_name), default=str))152    pitr_status = continuous_backup_json_response['ContinuousBackupsDescription']['PointInTimeRecoveryDescription']['PointInTimeRecoveryStatus']153    logging.info("PointInTimeRecoveryStatus: %s", pitr_status)154    return pitr_status155def get_earliest_restorable_point(table_name, aws_profile):156    session = boto3.Session(profile_name=aws_profile)157    dynamodb_client = session.client('dynamodb')158    continuous_backup_json_response = json.loads(json.dumps(dynamodb_client.describe_continuous_backups(TableName=table_name), default=str))159    earliest_restorable_datetime = continuous_backup_json_response['ContinuousBackupsDescription']['PointInTimeRecoveryDescription']['EarliestRestorableDateTime']160    logging.info("EarliestRestorableDateTime: %s", earliest_restorable_datetime)161    return earliest_restorable_datetime162def recover_lro_store(source_client, dest_client, restore_datetime, env, aws_profile, seconds_to_add_in_erp = 0):163    try:164        source_table = utils.get_lro_store_table_name(source_client, env)165        target_table = utils.get_lro_store_table_name(dest_client, env)166        is_table_exist(source_table, aws_profile)167        source_table_pitr_status = get_pitr_status(source_table, aws_profile)168        if source_table_pitr_status == "DISABLED":169            logging.info("Source table %s Point in time recovery is DISABLED, so we can't proceed with recovery", source_table)170            raise Exception("Source table PITR is Disabled!")171        restore_datetime = datetime.strptime(restore_datetime, "%Y-%m-%d %H:%M:%S")172        restore_datetime_utc = restore_datetime.replace(tzinfo=timezone.utc)173        log_arguments(source_client, dest_client, aws_profile,  restore_datetime_utc.timestamp(), env)174        earliest_restorable_point = utils.convert_datetime_to_utc_tz(get_earliest_restorable_point(source_table, aws_profile)) + timedelta(seconds=seconds_to_add_in_erp)175        truncate_table_required = False176        if (restore_datetime_utc > earliest_restorable_point):177            logging.info("Recovery point %s is after earliest restorable point %s", restore_datetime_utc, earliest_restorable_point)178            actual_restoration_point = restore_datetime_utc.timestamp()179        else:180            logging.info("Recovery point %s is before earliest restorable point %s", restore_datetime_utc, earliest_restorable_point)181            actual_restoration_point = earliest_restorable_point.timestamp()182            truncate_table_required =True183        delete_table_if_exist(target_table, aws_profile)184        restore_table_to_point_in_time(source_table, target_table, str(actual_restoration_point))185        wait_for_table_to_be_in_active_status(target_table)186        enable_point_in_time_recovery_on_table(target_table, aws_profile)187        if truncate_table_required:188            delete_records_with_last_updated_time_after_recovery_point(target_table, restore_datetime, aws_profile)189        logging.info("Point in Time Recovery is successfully done! in table %s", target_table)190        return 0191    except Exception:192        return 1193def enable_pitr(table_name, aws_profile):194    enable_point_in_time_recovery_on_table(table_name, aws_profile)195if __name__ == "__main__":196    args = get_args_parser().parse_args()197    sys.exit(recover_lro_store(args.source_client_name, args.target_client_name, args.restore_datetime, args.env, args.aws_profile))198

dynamodb_restore.py

Source:dynamodb_restore.py

...51                kwargs["UseLatestRestorableTime"] = True52                kwargs.pop("RestoreDateTime", None)53            else:54                kwargs["RestoreDateTime"] = self.restore_datetime55        self.ddb_low_level.client.restore_table_to_point_in_time(56            SourceTableName=self.full_table_name,57            TargetTableName=self.aws_restored_table,58            **kwargs,59        )60    def wait_for_aws_restored_table_ready(self):61        self.ddb_client.wait_until_table_exists(62            table_name=self.aws_restored_table, table_name_verbatim=True63        )64    def create_local_dump_of_aws_restored_table(self):65        """66        Dynamodump was removed as a dependency in setup.py to prevent conflicts with67        recent version of boto3. To install it, use "pip install dynamodump"68        """69        subprocess.run(...

pitr.py

Source:pitr.py

...42        latest_date = response['ContinuousBackupsDescription']['PointInTimeRecoveryDescription']['LatestRestorableDateTime']43        return True, earliest_date, latest_date44    return False45def restore_table(table_name, target_table_name, point_in_time):46    response = dynamodb.restore_table_to_point_in_time(47        SourceTableName=table_name,48        TargetTableName=target_table_name,49        RestoreDateTime=point_in_time,50        UseLatestRestorableTime=False51    )52    return response53def delete_table(table_name):54    response = dynamodb.delete_table(55        TableName=table_name56    )57    return response58def handler(event, context):59    response = None60    date_string_format = '%Y-%m-%d %H:%M:%S'...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.