<s> import boto . dynamodb2 . layer1 , json , sys , time , shutil , os , argparse , logging , datetime , threading <EOL> from boto . dynamodb2 . layer1 import DynamoDBConnection <EOL> JSON_INDENT = <NUM_LIT:2> <EOL> AWS_SLEEP_INTERVAL = <NUM_LIT:10> <EOL> LOCAL_SLEEP_INTERVAL = <NUM_LIT:1> <EOL> MAX_BATCH_WRITE = <NUM_LIT> <EOL> SCHEMA_FILE = "<STR_LIT>" <EOL> DATA_DIR = "<STR_LIT:data>" <EOL> MAX_RETRY = <NUM_LIT:6> <EOL> LOCAL_REGION = "<STR_LIT>" <EOL> LOG_LEVEL = "<STR_LIT>" <EOL> DUMP_PATH = "<STR_LIT>" <EOL> RESTORE_WRITE_CAPACITY = <NUM_LIT> <EOL> THREAD_START_DELAY = <NUM_LIT:1> <EOL> CURRENT_WORKING_DIR = os . getcwd ( ) <EOL> DEFAULT_PREFIX_SEPARATOR = "<STR_LIT:->" <EOL> def get_table_name_matches ( conn , table_name_wildcard , separator ) : <EOL> all_tables = [ ] <EOL> last_evaluated_table_name = None <EOL> while True : <EOL> table_list = conn . list_tables ( exclusive_start_table_name = last_evaluated_table_name ) <EOL> all_tables . extend ( table_list [ "<STR_LIT>" ] ) <EOL> try : <EOL> last_evaluated_table_name = table_list [ "<STR_LIT>" ] <EOL> except KeyError , e : <EOL> break <EOL> matching_tables = [ ] <EOL> for table_name in all_tables : <EOL> if table_name_wildcard == "<STR_LIT:*>" : <EOL> matching_tables . append ( table_name ) <EOL> elif separator == None : <EOL> if table_name . startswith ( table_name_wildcard . split ( "<STR_LIT:*>" , <NUM_LIT:1> ) [ <NUM_LIT:0> ] ) : <EOL> matching_tables . append ( table_name ) <EOL> elif table_name . split ( separator , <NUM_LIT:1> ) [ <NUM_LIT:0> ] == table_name_wildcard . split ( "<STR_LIT:*>" , <NUM_LIT:1> ) [ <NUM_LIT:0> ] : <EOL> matching_tables . append ( table_name ) <EOL> return matching_tables <EOL> def get_restore_table_matches ( table_name_wildcard , separator ) : <EOL> matching_tables = [ ] <EOL> try : <EOL> dir_list = os . listdir ( "<STR_LIT>" + DUMP_PATH ) <EOL> except OSError : <EOL> logging . info ( "<STR_LIT>" % DUMP_PATH ) <EOL> dump_data_path = CURRENT_WORKING_DIR <EOL> try : <EOL> dir_list = os . listdir ( dump_data_path ) <EOL> except OSError : <EOL> logging . info ( "<STR_LIT>" % dump_data_path ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> for dir_name in dir_list : <EOL> if table_name_wildcard == "<STR_LIT:*>" : <EOL> matching_tables . append ( dir_name ) <EOL> elif dir_name . split ( separator , <NUM_LIT:1> ) [ <NUM_LIT:0> ] == table_name_wildcard . split ( "<STR_LIT:*>" , <NUM_LIT:1> ) [ <NUM_LIT:0> ] : <EOL> matching_tables . append ( dir_name ) <EOL> return matching_tables <EOL> def change_prefix ( source_table_name , source_wildcard , destination_wildcard , separator ) : <EOL> source_prefix = source_wildcard . split ( "<STR_LIT:*>" , <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> destination_prefix = destination_wildcard . split ( "<STR_LIT:*>" , <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> if source_table_name . split ( separator , <NUM_LIT:1> ) [ <NUM_LIT:0> ] == source_prefix : <EOL> return destination_prefix + separator + source_table_name . split ( separator , <NUM_LIT:1> ) [ <NUM_LIT:1> ] <EOL> def delete_table ( conn , sleep_interval , table_name ) : <EOL> if not args . dataOnly : <EOL> while True : <EOL> table_exist = True <EOL> try : <EOL> conn . delete_table ( table_name ) <EOL> except boto . exception . JSONResponseError , e : <EOL> if e . body [ "<STR_LIT>" ] == "<STR_LIT>" : <EOL> table_exist = False <EOL> logging . info ( table_name + "<STR_LIT>" ) <EOL> break <EOL> elif e . body [ "<STR_LIT>" ] == "<STR_LIT>" : <EOL> logging . info ( "<STR_LIT>" + table_name + "<STR_LIT:..>" ) <EOL> time . sleep ( sleep_interval ) <EOL> elif e . body [ "<STR_LIT>" ] == "<STR_LIT>" : <EOL> logging . info ( "<STR_LIT>" + table_name + "<STR_LIT:..>" ) <EOL> time . sleep ( sleep_interval ) <EOL> elif e . body [ "<STR_LIT>" ] == "<STR_LIT>" : <EOL> logging . info ( table_name + "<STR_LIT>" ) <EOL> time . sleep ( sleep_interval ) <EOL> else : <EOL> logging . exception ( e ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> if table_exist : <EOL> try : <EOL> while True : <EOL> logging . info ( "<STR_LIT>" + table_name + "<STR_LIT>" + conn . describe_table ( table_name ) [ "<STR_LIT>" ] [ "<STR_LIT>" ] + "<STR_LIT:]>" ) <EOL> time . sleep ( sleep_interval ) <EOL> except boto . exception . JSONResponseError , e : <EOL> if e . body [ "<STR_LIT>" ] == "<STR_LIT>" : <EOL> logging . info ( table_name + "<STR_LIT>" ) <EOL> pass <EOL> else : <EOL> logging . exception ( e ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> def mkdir_p ( path ) : <EOL> try : <EOL> os . makedirs ( path ) <EOL> except OSError as exc : <EOL> if exc . errno == errno . EEXIST and os . path . isdir ( path ) : <EOL> pass <EOL> else : raise <EOL> def batch_write ( conn , sleep_interval , table_name , put_requests ) : <EOL> request_items = { table_name : put_requests } <EOL> i = <NUM_LIT:1> <EOL> while True : <EOL> response = conn . batch_write_item ( request_items ) <EOL> unprocessed_items = response [ "<STR_LIT>" ] <EOL> if len ( unprocessed_items ) == <NUM_LIT:0> : <EOL> break <EOL> if len ( unprocessed_items ) > <NUM_LIT:0> and i <= MAX_RETRY : <EOL> logging . debug ( str ( len ( unprocessed_items ) ) + "<STR_LIT>" + str ( i ) + "<STR_LIT:]>" ) <EOL> request_items = unprocessed_items <EOL> i += <NUM_LIT:1> <EOL> else : <EOL> logging . info ( "<STR_LIT>" + json . dumps ( unprocessed_items , indent = JSON_INDENT ) ) <EOL> logging . info ( "<STR_LIT>" ) <EOL> break <EOL> def wait_for_active_table ( conn , table_name , verb ) : <EOL> while True : <EOL> if conn . describe_table ( table_name ) [ "<STR_LIT>" ] [ "<STR_LIT>" ] != "<STR_LIT>" : <EOL> logging . info ( "<STR_LIT>" + table_name + "<STR_LIT>" + verb + "<STR_LIT>" + conn . describe_table ( table_name ) [ "<STR_LIT>" ] [ "<STR_LIT>" ] + "<STR_LIT:]>" ) <EOL> time . sleep ( sleep_interval ) <EOL> else : <EOL> logging . info ( table_name + "<STR_LIT:U+0020>" + verb + "<STR_LIT:.>" ) <EOL> break <EOL> def update_provisioned_throughput ( conn , table_name , read_capacity , write_capacity , wait = True ) : <EOL> logging . info ( "<STR_LIT>" + table_name + "<STR_LIT>" + str ( read_capacity ) + "<STR_LIT>" + str ( write_capacity ) ) <EOL> while True : <EOL> try : <EOL> conn . update_table ( table_name , { "<STR_LIT>" : int ( read_capacity ) , "<STR_LIT>" : int ( write_capacity ) } ) <EOL> break <EOL> except boto . exception . JSONResponseError , e : <EOL> if e . body [ "<STR_LIT>" ] == "<STR_LIT>" : <EOL> logging . info ( "<STR_LIT>" + table_name + "<STR_LIT:..>" ) <EOL> time . sleep ( sleep_interval ) <EOL> elif e . body [ "<STR_LIT>" ] == "<STR_LIT>" : <EOL> logging . info ( "<STR_LIT>" + table_name + "<STR_LIT:..>" ) <EOL> time . sleep ( sleep_interval ) <EOL> if wait : <EOL> wait_for_active_table ( conn , table_name , "<STR_LIT>" ) <EOL> def do_empty ( conn , table_name ) : <EOL> logging . info ( "<STR_LIT>" + table_name + "<STR_LIT:..>" ) <EOL> logging . info ( "<STR_LIT>" + table_name ) <EOL> table_data = conn . describe_table ( table_name ) <EOL> table_desc = table_data [ "<STR_LIT>" ] <EOL> table_attribute_definitions = table_desc [ "<STR_LIT>" ] <EOL> table_key_schema = table_desc [ "<STR_LIT>" ] <EOL> original_read_capacity = table_desc [ "<STR_LIT>" ] [ "<STR_LIT>" ] <EOL> original_write_capacity = table_desc [ "<STR_LIT>" ] [ "<STR_LIT>" ] <EOL> table_local_secondary_indexes = table_desc . get ( "<STR_LIT>" ) <EOL> table_global_secondary_indexes = table_desc . get ( "<STR_LIT>" ) <EOL> table_provisioned_throughput = { "<STR_LIT>" : int ( original_read_capacity ) , "<STR_LIT>" : int ( original_write_capacity ) } <EOL> logging . info ( "<STR_LIT>" + table_name ) <EOL> delete_table ( conn , sleep_interval , table_name ) <EOL> logging . info ( "<STR_LIT>" + table_name ) <EOL> while True : <EOL> try : <EOL> conn . create_table ( table_attribute_definitions , table_name , table_key_schema , table_provisioned_throughput , table_local_secondary_indexes , table_global_secondary_indexes ) <EOL> break <EOL> except boto . exception . JSONResponseError , e : <EOL> if e . body [ "<STR_LIT>" ] == "<STR_LIT>" : <EOL> logging . info ( "<STR_LIT>" + destination_table + "<STR_LIT:..>" ) <EOL> time . sleep ( sleep_interval ) <EOL> elif e . body [ "<STR_LIT>" ] == "<STR_LIT>" : <EOL> logging . info ( "<STR_LIT>" + destination_table + "<STR_LIT:..>" ) <EOL> time . sleep ( sleep_interval ) <EOL> else : <EOL> logging . exception ( e ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> wait_for_active_table ( conn , table_name , "<STR_LIT>" ) <EOL> logging . info ( "<STR_LIT>" + table_name + "<STR_LIT>" + str ( datetime . datetime . now ( ) . replace ( microsecond = <NUM_LIT:0> ) - start_time ) ) <EOL> def do_backup ( conn , table_name , read_capacity ) : <EOL> logging . info ( "<STR_LIT>" + table_name + "<STR_LIT:..>" ) <EOL> if os . path . exists ( DUMP_PATH + "<STR_LIT:/>" + table_name ) : <EOL> shutil . rmtree ( DUMP_PATH + "<STR_LIT:/>" + table_name ) <EOL> mkdir_p ( DUMP_PATH + "<STR_LIT:/>" + table_name ) <EOL> logging . info ( "<STR_LIT>" + table_name ) <EOL> f = open ( DUMP_PATH + "<STR_LIT:/>" + table_name + "<STR_LIT:/>" + SCHEMA_FILE , "<STR_LIT>" ) <EOL> table_desc = conn . describe_table ( table_name ) <EOL> f . write ( json . dumps ( table_desc , indent = JSON_INDENT ) ) <EOL> f . close ( ) <EOL> original_read_capacity = table_desc [ "<STR_LIT>" ] [ "<STR_LIT>" ] [ "<STR_LIT>" ] <EOL> original_write_capacity = table_desc [ "<STR_LIT>" ] [ "<STR_LIT>" ] [ "<STR_LIT>" ] <EOL> if read_capacity != None and read_capacity != original_read_capacity : <EOL> update_provisioned_throughput ( conn , table_name , read_capacity , original_write_capacity ) <EOL> logging . info ( "<STR_LIT>" + table_name ) <EOL> mkdir_p ( DUMP_PATH + "<STR_LIT:/>" + table_name + "<STR_LIT:/>" + DATA_DIR ) <EOL> i = <NUM_LIT:1> <EOL> last_evaluated_key = None <EOL> while True : <EOL> scanned_table = conn . scan ( table_name , exclusive_start_key = last_evaluated_key ) <EOL> f = open ( DUMP_PATH + "<STR_LIT:/>" + table_name + "<STR_LIT:/>" + DATA_DIR + "<STR_LIT:/>" + str ( i ) . zfill ( <NUM_LIT:4> ) + "<STR_LIT>" , "<STR_LIT>" ) <EOL> f . write ( json . dumps ( scanned_table , indent = JSON_INDENT ) ) <EOL> f . close ( ) <EOL> i += <NUM_LIT:1> <EOL> try : <EOL> last_evaluated_key = scanned_table [ "<STR_LIT>" ] <EOL> except KeyError , e : <EOL> break <EOL> if read_capacity != None and read_capacity != original_read_capacity : <EOL> update_provisioned_throughput ( conn , table_name , original_read_capacity , original_write_capacity , False ) <EOL> logging . info ( "<STR_LIT>" + table_name + "<STR_LIT>" + str ( datetime . datetime . now ( ) . replace ( microsecond = <NUM_LIT:0> ) - start_time ) ) <EOL> def do_restore ( conn , sleep_interval , source_table , destination_table , write_capacity ) : <EOL> logging . info ( "<STR_LIT>" + source_table + "<STR_LIT>" + destination_table + "<STR_LIT:..>" ) <EOL> if os . path . exists ( "<STR_LIT>" % ( DUMP_PATH , source_table ) ) : <EOL> dump_data_path = DUMP_PATH <EOL> else : <EOL> logging . info ( "<STR_LIT>" % ( DUMP_PATH , source_table ) ) <EOL> if os . path . exists ( "<STR_LIT>" % ( CURRENT_WORKING_DIR , source_table ) ) : <EOL> dump_data_path = CURRENT_WORKING_DIR <EOL> else : <EOL> logging . info ( "<STR_LIT>" % ( CURRENT_WORKING_DIR , source_table ) ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> table_data = json . load ( open ( dump_data_path + "<STR_LIT:/>" + source_table + "<STR_LIT:/>" + SCHEMA_FILE ) ) <EOL> table = table_data [ "<STR_LIT>" ] <EOL> table_attribute_definitions = table [ "<STR_LIT>" ] <EOL> table_table_name = destination_table <EOL> table_key_schema = table [ "<STR_LIT>" ] <EOL> original_read_capacity = table [ "<STR_LIT>" ] [ "<STR_LIT>" ] <EOL> original_write_capacity = table [ "<STR_LIT>" ] [ "<STR_LIT>" ] <EOL> table_local_secondary_indexes = table . get ( "<STR_LIT>" ) <EOL> table_global_secondary_indexes = table . get ( "<STR_LIT>" ) <EOL> if not args . dataOnly : <EOL> if write_capacity == None : <EOL> if original_write_capacity < RESTORE_WRITE_CAPACITY : <EOL> write_capacity = RESTORE_WRITE_CAPACITY <EOL> else : <EOL> write_capacity = original_write_capacity <EOL> original_gsi_write_capacities = [ ] <EOL> if table_global_secondary_indexes is not None : <EOL> for gsi in table_global_secondary_indexes : <EOL> original_gsi_write_capacities . append ( gsi [ "<STR_LIT>" ] [ "<STR_LIT>" ] ) <EOL> if gsi [ "<STR_LIT>" ] [ "<STR_LIT>" ] < int ( write_capacity ) : <EOL> gsi [ "<STR_LIT>" ] [ "<STR_LIT>" ] = int ( write_capacity ) <EOL> table_provisioned_throughput = { "<STR_LIT>" : int ( original_read_capacity ) , "<STR_LIT>" : int ( write_capacity ) } <EOL> logging . info ( "<STR_LIT>" + destination_table + "<STR_LIT>" + str ( write_capacity ) ) <EOL> while True : <EOL> try : <EOL> conn . create_table ( table_attribute_definitions , table_table_name , table_key_schema , table_provisioned_throughput , table_local_secondary_indexes , table_global_secondary_indexes ) <EOL> break <EOL> except boto . exception . JSONResponseError , e : <EOL> if e . body [ "<STR_LIT>" ] == "<STR_LIT>" : <EOL> logging . info ( "<STR_LIT>" + destination_table + "<STR_LIT:..>" ) <EOL> time . sleep ( sleep_interval ) <EOL> elif e . body [ "<STR_LIT>" ] == "<STR_LIT>" : <EOL> logging . info ( "<STR_LIT>" + destination_table + "<STR_LIT:..>" ) <EOL> time . sleep ( sleep_interval ) <EOL> else : <EOL> logging . exception ( e ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> wait_for_active_table ( conn , destination_table , "<STR_LIT>" ) <EOL> logging . info ( "<STR_LIT>" + destination_table + "<STR_LIT>" ) <EOL> data_file_list = os . listdir ( dump_data_path + "<STR_LIT:/>" + source_table + "<STR_LIT:/>" + DATA_DIR + "<STR_LIT:/>" ) <EOL> data_file_list . sort ( ) <EOL> for data_file in data_file_list : <EOL> logging . info ( "<STR_LIT>" + data_file + "<STR_LIT>" + destination_table ) <EOL> items = [ ] <EOL> item_data = json . load ( open ( dump_data_path + "<STR_LIT:/>" + source_table + "<STR_LIT:/>" + DATA_DIR + "<STR_LIT:/>" + data_file ) ) <EOL> items . extend ( item_data [ "<STR_LIT>" ] ) <EOL> put_requests = [ ] <EOL> while len ( items ) > <NUM_LIT:0> : <EOL> put_requests . append ( { "<STR_LIT>" : { "<STR_LIT>" : items . pop ( <NUM_LIT:0> ) } } ) <EOL> if len ( put_requests ) == MAX_BATCH_WRITE : <EOL> logging . debug ( "<STR_LIT>" + str ( MAX_BATCH_WRITE ) + "<STR_LIT>" + destination_table + "<STR_LIT:..>" ) <EOL> batch_write ( conn , sleep_interval , destination_table , put_requests ) <EOL> del put_requests [ : ] <EOL> if len ( put_requests ) > <NUM_LIT:0> : <EOL> batch_write ( conn , sleep_interval , destination_table , put_requests ) <EOL> if not args . dataOnly and not args . skipThroughputUpdate : <EOL> if write_capacity != original_write_capacity : <EOL> update_provisioned_throughput ( conn , destination_table , original_read_capacity , original_write_capacity , False ) <EOL> if table_global_secondary_indexes is not None : <EOL> gsi_data = [ ] <EOL> for gsi in table_global_secondary_indexes : <EOL> original_gsi_write_capacity = original_gsi_write_capacities . pop ( <NUM_LIT:0> ) <EOL> if original_gsi_write_capacity != gsi [ "<STR_LIT>" ] [ "<STR_LIT>" ] : <EOL> gsi_data . append ( { "<STR_LIT>" : { "<STR_LIT>" : gsi [ "<STR_LIT>" ] , "<STR_LIT>" : { "<STR_LIT>" : int ( gsi [ "<STR_LIT>" ] [ "<STR_LIT>" ] ) , "<STR_LIT>" : int ( original_gsi_write_capacity ) , } , } , } ) <EOL> logging . info ( "<STR_LIT>" + destination_table + "<STR_LIT>" ) <EOL> while True : <EOL> try : <EOL> conn . update_table ( destination_table , global_secondary_index_updates = gsi_data ) <EOL> break <EOL> except boto . exception . JSONResponseError , e : <EOL> if e . body [ "<STR_LIT>" ] == "<STR_LIT>" : <EOL> logging . info ( "<STR_LIT>" + destination_table + "<STR_LIT:..>" ) <EOL> time . sleep ( sleep_interval ) <EOL> elif e . body [ "<STR_LIT>" ] == "<STR_LIT>" : <EOL> logging . info ( "<STR_LIT>" + destination_table + "<STR_LIT:..>" ) <EOL> time . sleep ( sleep_interval ) <EOL> logging . info ( "<STR_LIT>" + source_table + "<STR_LIT>" + destination_table + "<STR_LIT>" + str ( datetime . datetime . now ( ) . replace ( microsecond = <NUM_LIT:0> ) - start_time ) ) <EOL> parser = argparse . ArgumentParser ( description = "<STR_LIT>" ) <EOL> parser . add_argument ( "<STR_LIT>" , "<STR_LIT>" , help = "<STR_LIT>" ) <EOL> parser . add_argument ( "<STR_LIT>" , "<STR_LIT>" , help = "<STR_LIT>" + LOCAL_REGION + "<STR_LIT>" ) <EOL> parser . add_argument ( "<STR_LIT>" , "<STR_LIT>" , help = "<STR_LIT>" ) <EOL> parser . add_argument ( "<STR_LIT>" , "<STR_LIT>" , help = "<STR_LIT>" ) <EOL> parser . add_argument ( "<STR_LIT>" , help = "<STR_LIT>" ) <EOL> parser . add_argument ( "<STR_LIT>" , action = '<STR_LIT:store_true>' , help = "<STR_LIT>" ) <EOL> parser . add_argument ( "<STR_LIT>" , help = "<STR_LIT>" ) <EOL> parser . add_argument ( "<STR_LIT>" , help = "<STR_LIT>" + str ( RESTORE_WRITE_CAPACITY ) + "<STR_LIT>" ) <EOL> parser . add_argument ( "<STR_LIT>" , help = "<STR_LIT>" ) <EOL> parser . add_argument ( "<STR_LIT>" , help = "<STR_LIT>" ) <EOL> parser . add_argument ( "<STR_LIT>" , help = "<STR_LIT>" ) <EOL> parser . add_argument ( "<STR_LIT>" , help = "<STR_LIT>" ) <EOL> parser . add_argument ( "<STR_LIT>" , help = "<STR_LIT>" ) <EOL> parser . add_argument ( "<STR_LIT>" , action = "<STR_LIT:store_true>" , default = False , help = "<STR_LIT>" ) <EOL> parser . add_argument ( "<STR_LIT>" , action = "<STR_LIT:store_true>" , default = False , help = "<STR_LIT>" ) <EOL> args = parser . parse_args ( ) <EOL> log_level = LOG_LEVEL <EOL> if args . log != None : <EOL> log_level = args . log . upper ( ) <EOL> logging . basicConfig ( level = getattr ( logging , log_level ) ) <EOL> if args . region == LOCAL_REGION : <EOL> conn = DynamoDBConnection ( aws_access_key_id = args . accessKey , aws_secret_access_key = args . secretKey , host = args . host , port = int ( args . port ) , is_secure = False ) <EOL> sleep_interval = LOCAL_SLEEP_INTERVAL <EOL> else : <EOL> conn = boto . dynamodb2 . connect_to_region ( args . region , aws_access_key_id = args . accessKey , aws_secret_access_key = args . secretKey ) <EOL> sleep_interval = AWS_SLEEP_INTERVAL <EOL> prefix_separator = DEFAULT_PREFIX_SEPARATOR <EOL> if args . prefixSeparator != None : <EOL> prefix_separator = args . prefixSeparator <EOL> if args . noSeparator == True : <EOL> prefix_separator = None <EOL> start_time = datetime . datetime . now ( ) . replace ( microsecond = <NUM_LIT:0> ) <EOL> if args . mode == "<STR_LIT>" : <EOL> if args . srcTable . find ( "<STR_LIT:*>" ) != - <NUM_LIT:1> : <EOL> matching_backup_tables = get_table_name_matches ( conn , args . srcTable , prefix_separator ) <EOL> logging . info ( "<STR_LIT>" + str ( len ( matching_backup_tables ) ) + "<STR_LIT>" + "<STR_LIT:U+002CU+0020>" . join ( matching_backup_tables ) ) <EOL> threads = [ ] <EOL> for table_name in matching_backup_tables : <EOL> t = threading . Thread ( target = do_backup , args = ( conn , table_name , args . readCapacity , ) ) <EOL> threads . append ( t ) <EOL> t . start ( ) <EOL> time . sleep ( THREAD_START_DELAY ) <EOL> for thread in threads : <EOL> thread . join ( ) <EOL> logging . info ( "<STR_LIT>" + args . srcTable + "<STR_LIT>" ) <EOL> else : <EOL> do_backup ( conn , args . srcTable , args . readCapacity ) <EOL> elif args . mode == "<STR_LIT>" : <EOL> if args . destTable != None : <EOL> dest_table = args . destTable <EOL> else : <EOL> dest_table = args . srcTable <EOL> if dest_table . find ( "<STR_LIT:*>" ) != - <NUM_LIT:1> : <EOL> matching_destination_tables = get_table_name_matches ( conn , dest_table , prefix_separator ) <EOL> delete_str = "<STR_LIT>" if args . dataOnly else "<STR_LIT>" <EOL> logging . info ( "<STR_LIT>" + str ( len ( matching_destination_tables ) ) + "<STR_LIT>" + delete_str + "<STR_LIT:U+002CU+0020>" . join ( matching_destination_tables ) ) <EOL> threads = [ ] <EOL> for table_name in matching_destination_tables : <EOL> t = threading . Thread ( target = delete_table , args = ( conn , sleep_interval , table_name , ) ) <EOL> threads . append ( t ) <EOL> t . start ( ) <EOL> time . sleep ( THREAD_START_DELAY ) <EOL> for thread in threads : <EOL> thread . join ( ) <EOL> matching_restore_tables = get_restore_table_matches ( args . srcTable , prefix_separator ) <EOL> logging . info ( "<STR_LIT>" + str ( len ( matching_restore_tables ) ) + "<STR_LIT>" + DUMP_PATH + "<STR_LIT>" + "<STR_LIT:U+002CU+0020>" . join ( matching_restore_tables ) ) <EOL> threads = [ ] <EOL> for source_table in matching_restore_tables : <EOL> if args . srcTable == "<STR_LIT:*>" : <EOL> t = threading . Thread ( target = do_restore , args = ( conn , sleep_interval , source_table , source_table , args . writeCapacity ) ) <EOL> else : <EOL> t = threading . Thread ( target = do_restore , args = ( conn , sleep_interval , source_table , change_prefix ( source_table , args . srcTable , dest_table , prefix_separator ) , args . writeCapacity , ) ) <EOL> threads . append ( t ) <EOL> t . start ( ) <EOL> time . sleep ( THREAD_START_DELAY ) <EOL> for thread in threads : <EOL> thread . join ( ) <EOL> logging . info ( "<STR_LIT>" + args . srcTable + "<STR_LIT>" + dest_table + "<STR_LIT>" ) <EOL> else : <EOL> delete_table ( conn , sleep_interval , dest_table ) <EOL> do_restore ( conn , sleep_interval , args . srcTable , dest_table , args . writeCapacity ) <EOL> elif args . mode == "<STR_LIT>" : <EOL> if args . srcTable . find ( "<STR_LIT:*>" ) != - <NUM_LIT:1> : <EOL> matching_backup_tables = get_table_name_matches ( conn , args . srcTable , prefix_separator ) <EOL> logging . info ( "<STR_LIT>" + str ( len ( matching_backup_tables ) ) + "<STR_LIT>" + "<STR_LIT:U+002CU+0020>" . join ( matching_backup_tables ) ) <EOL> threads = [ ] <EOL> for table_name in matching_backup_tables : <EOL> t = threading . Thread ( target = do_empty , args = ( conn , table_name ) ) <EOL> threads . append ( t ) <EOL> t . start ( ) <EOL> time . sleep ( THREAD_START_DELAY ) <EOL> for thread in threads : <EOL> thread . join ( ) <EOL> logging . info ( "<STR_LIT>" + args . srcTable + "<STR_LIT>" ) <EOL> else : <EOL> do_empty ( conn , args . srcTable ) </s>
