()
Entrypoint to the script
Entrypoint to the script
Entrypoint to the script
def main():
    """
    Entrypoint to the script
    """

    global args, sleep_interval, start_time

    # parse args
    parser = argparse.ArgumentParser(
        description="Simple DynamoDB backup/restore/empty."
    )
    parser.add_argument(
        "-a",
        "--archive",
        help="Type of compressed archive to create." "If unset, don't create archive",
        choices=["zip", "tar"],
    )
    parser.add_argument(
        "-b",
        "--bucket",
        help="S3 bucket in which to store or retrieve backups." "[must already exist]",
    )
    parser.add_argument(
        "-m",
        "--mode",
        help="Operation to perform",
        choices=["backup", "restore", "empty"],
    )
    parser.add_argument(
        "-r",
        "--region",
        help="AWS region to use, e.g. 'us-west-1'. "
        "Can use AWS_DEFAULT_REGION for local testing.  Use '"
        + LOCAL_REGION
        + "' for local DynamoDB testing",
    )
    parser.add_argument(
        "--host", help="Host of local DynamoDB [required only for local]"
    )
    parser.add_argument(
        "--port", help="Port of local DynamoDB [required only for local]"
    )
    parser.add_argument(
        "--accessKey", help="Access key of local DynamoDB " "[required only for local]"
    )
    parser.add_argument(
        "--secretKey", help="Secret key of local DynamoDB " "[required only for local]"
    )
    parser.add_argument(
        "-p",
        "--profile",
        help="AWS credentials file profile to use. Allows you to use a "
        "profile instead accessKey, secretKey authentication",
    )
    parser.add_argument(
        "-s",
        "--srcTable",
        help="Source DynamoDB table name to backup or restore from, "
        "use 'tablename*' for wildcard prefix selection or '*' for "
        "all tables.  Mutually exclusive with --tag",
    )
    parser.add_argument(
        "-d",
        "--destTable",
        help="Destination DynamoDB table name to backup or restore to, "
        "use 'tablename*' for wildcard prefix selection "
        "(defaults to use '-' separator) [optional, defaults to source]",
    )
    parser.add_argument(
        "--prefixSeparator",
        help="Specify a different prefix separator, " "e.g. '.' [optional]",
    )
    parser.add_argument(
        "--noSeparator",
        action="store_true",
        help="Overrides the use of a prefix separator for backup wildcard "
        "searches [optional]",
    )
    parser.add_argument(
        "--readCapacity",
        help="Change the temp read capacity of the DynamoDB table to backup "
        "from [optional]",
    )
    parser.add_argument(
        "-t",
        "--tag",
        help="Tag to use for identifying tables to back up.  "
        "Mutually exclusive with srcTable.  Provided as KEY=VALUE",
    )
    parser.add_argument(
        "--writeCapacity",
        help="Change the temp write capacity of the DynamoDB table to restore "
        "to [defaults to " + str(RESTORE_WRITE_CAPACITY) + ", optional]",
    )
    parser.add_argument(
        "--schemaOnly",
        action="store_true",
        default=False,
        help="Backup or restore the schema only. Do not backup/restore data. "
        "Can be used with both backup and restore modes. Cannot be used with "
        "the --dataOnly [optional]",
    )
    parser.add_argument(
        "--dataOnly",
        action="store_true",
        default=False,
        help="Restore data only. Do not delete/recreate schema [optional for "
        "restore]",
    )
    parser.add_argument(
        "--noConfirm",
        action="store_true",
        default=False,
        help="Don't ask for confirmation before deleting existing schemas.",
    )
    parser.add_argument(
        "--skipThroughputUpdate",
        action="store_true",
        default=False,
        help="Skip updating throughput values across tables [optional]",
    )
    parser.add_argument(
        "--dumpPath",
        help="Directory to place and search for DynamoDB table "
        "backups (defaults to use '" + str(DATA_DUMP) + "') [optional]",
        default=str(DATA_DUMP),
    )
    parser.add_argument(
        "--billingMode",
        help="Set billing mode between "
        + str(PROVISIONED_BILLING_MODE)
        + "|"
        + str(PAY_PER_REQUEST_BILLING_MODE)
        + " (defaults to use '"
        + str(PROVISIONED_BILLING_MODE)
        + "') [optional]",
        choices=[PROVISIONED_BILLING_MODE, PAY_PER_REQUEST_BILLING_MODE],
        default=str(PROVISIONED_BILLING_MODE),
    )
    parser.add_argument(
        "--log", help="Logging level - DEBUG|INFO|WARNING|ERROR|CRITICAL " "[optional]"
    )
    args = parser.parse_args()

    # set log level
    log_level = LOG_LEVEL
    if args.log is not None:
        log_level = args.log.upper()
    logging.basicConfig(level=getattr(logging, log_level))

    # Check to make sure that --dataOnly and --schemaOnly weren't simultaneously specified
    if args.schemaOnly and args.dataOnly:
        logging.info("Options --schemaOnly and --dataOnly are mutually exclusive.")
        sys.exit(1)

    # instantiate connection
    if args.region == LOCAL_REGION:
        conn = _get_aws_client(
            service="dynamodb",
            access_key=args.accessKey,
            secret_key=args.secretKey,
            region=args.region,
            endpoint_url="http://" + args.host + ":" + args.port,
        )
        sleep_interval = LOCAL_SLEEP_INTERVAL
    else:
        if not args.profile:
            conn = _get_aws_client(
                service="dynamodb",
                access_key=args.accessKey,
                secret_key=args.secretKey,
                region=args.region,
            )
            sleep_interval = AWS_SLEEP_INTERVAL
        else:
            conn = _get_aws_client(
                service="dynamodb",
                profile=args.profile,
                region=args.region,
            )
            sleep_interval = AWS_SLEEP_INTERVAL

    # don't proceed if connection is not established
    if not conn:
        logging.info("Unable to establish connection with dynamodb")
        sys.exit(1)

    # set prefix separator
    prefix_separator = DEFAULT_PREFIX_SEPARATOR
    if args.prefixSeparator is not None:
        prefix_separator = args.prefixSeparator
    if args.noSeparator is True:
        prefix_separator = None

    # do backup/restore
    start_time = datetime.datetime.now().replace(microsecond=0)
    if args.mode == "backup":
        matching_backup_tables = []
        if args.tag:
            # Use Boto3 to find tags.  Boto3 provides a paginator that makes searching ta
            matching_backup_tables = get_table_name_by_tag(
                args.profile, args.region, args.tag
            )
        elif args.srcTable.find("*") != -1:
            matching_backup_tables = get_table_name_matches(
                conn, args.srcTable, prefix_separator
            )
        elif args.srcTable:
            matching_backup_tables.append(args.srcTable)

        if len(matching_backup_tables) == 0:
            logging.info("No matching tables found.  Nothing to do.")
            sys.exit(0)
        else:
            logging.info(
                "Found "
                + str(len(matching_backup_tables))
                + " table(s) in DynamoDB host to backup: "
                + ", ".join(matching_backup_tables)
            )

        try:
            if args.srcTable.find("*") == -1:
                do_backup(conn, args.read_capacity, tableQueue=None)
            else:
                do_backup(conn, args.read_capacity, matching_backup_tables)
        except AttributeError:
            # Didn't specify srcTable if we get here

            q = Queue()
            threads = []

            for i in range(MAX_NUMBER_BACKUP_WORKERS):
                t = threading.Thread(
                    target=do_backup,
                    args=(conn, args.readCapacity),
                    kwargs={"tableQueue": q},
                )
                t.start()
                threads.append(t)
                time.sleep(THREAD_START_DELAY)

            for table in matching_backup_tables:
                q.put(table)

            q.join()

            for i in range(MAX_NUMBER_BACKUP_WORKERS):
                q.put(None)
            for t in threads:
                t.join()

            try:
                logging.info("Backup of table(s) " + args.srcTable + " completed!")
            except (NameError, TypeError):
                logging.info(
                    "Backup of table(s) "
                    + ", ".join(matching_backup_tables)
                    + " completed!"
                )

            if args.archive:
                if args.tag:
                    for table in matching_backup_tables:
                        dump_path = args.dumpPath + os.sep + table
                        did_archive, archive_file = do_archive(args.archive, dump_path)
                        if args.bucket and did_archive:
                            do_put_bucket_object(
                                args.profile, args.region, args.bucket, archive_file
                            )
                else:
                    did_archive, archive_file = do_archive(args.archive, args.dumpPath)

                if args.bucket and did_archive:
                    do_put_bucket_object(
                        args.profile, args.region, args.bucket, archive_file
                    )

    elif args.mode == "restore":
        if args.destTable is not None:
            dest_table = args.destTable
        else:
            dest_table = args.srcTable

        # If backups are in S3 download and extract the backup to use during restoration
        if args.bucket:
            do_get_s3_archive(
                args.profile, args.region, args.bucket, args.srcTable, args.archive
            )

        if dest_table.find("*") != -1:
            matching_destination_tables = get_table_name_matches(
                conn, dest_table, prefix_separator
            )
            delete_str = ": " if args.dataOnly else " to be deleted: "
            logging.info(
                "Found "
                + str(len(matching_destination_tables))
                + " table(s) in DynamoDB host"
                + delete_str
                + ", ".join(matching_destination_tables)
            )

            threads = []
            for table in matching_destination_tables:
                t = threading.Thread(
                    target=delete_table, args=(conn, sleep_interval, table)
                )
                threads.append(t)
                t.start()
                time.sleep(THREAD_START_DELAY)

            for thread in threads:
                thread.join()

            matching_restore_tables = get_restore_table_matches(
                args.srcTable, prefix_separator
            )
            logging.info(
                "Found "
                + str(len(matching_restore_tables))
                + " table(s) in "
                + args.dumpPath
                + " to restore: "
                + ", ".join(matching_restore_tables)
            )

            threads = []
            for source_table in matching_restore_tables:
                if args.srcTable == "*":
                    t = threading.Thread(
                        target=do_restore,
                        args=(
                            conn,
                            sleep_interval,
                            source_table,
                            source_table,
                            args.writeCapacity,
                            args.billingMode,
                        ),
                    )
                else:
                    t = threading.Thread(
                        target=do_restore,
                        args=(
                            conn,
                            sleep_interval,
                            source_table,
                            change_prefix(
                                source_table,
                                args.srcTable,
                                dest_table,
                                prefix_separator,
                            ),
                            args.writeCapacity,
                            args.billingMode,
                        ),
                    )
                threads.append(t)
                t.start()
                time.sleep(THREAD_START_DELAY)

            for thread in threads:
                thread.join()

            logging.info(
                "Restore of table(s) "
                + args.srcTable
                + " to "
                + dest_table
                + " completed!"
            )
        else:
            delete_table(
                conn=conn, sleep_interval=sleep_interval, table_name=dest_table
            )
            do_restore(
                dynamo=conn,
                sleep_interval=sleep_interval,
                source_table=args.srcTable,
                destination_table=dest_table,
                write_capacity=args.writeCapacity,
                billing_mode=args.billingMode,
            )
    elif args.mode == "empty":
        if args.srcTable.find("*") != -1:
            matching_backup_tables = get_table_name_matches(
                conn, args.srcTable, prefix_separator
            )
            logging.info(
                "Found "
                + str(len(matching_backup_tables))
                + " table(s) in DynamoDB host to empty: "
                + ", ".join(matching_backup_tables)
            )

            threads = []
            for table in matching_backup_tables:
                t = threading.Thread(
                    target=do_empty, args=(conn, table, args.billingMode)
                )
                threads.append(t)
                t.start()
                time.sleep(THREAD_START_DELAY)

            for thread in threads:
                thread.join()

            logging.info("Empty of table(s) " + args.srcTable + " completed!")
        else:
            do_empty(conn, args.srcTable, args.billingMode)
def main ( ) : global args , sleep_interval , start_time # parse args parser = argparse . ArgumentParser ( description = "Simple DynamoDB backup/restore/empty." ) parser . add_argument ( "-a" , "--archive" , help = "Type of compressed archive to create." "If unset, don't create archive" , choices = [ "zip" , "tar" ] , ) parser . add_argument ( "-b" , "--bucket" , help = "S3 bucket in which to store or retrieve backups." "[must already exist]" , ) parser . add_argument ( "-m" , "--mode" , help = "Operation to perform" , choices = [ "backup" , "restore" , "empty" ] , ) parser . add_argument ( "-r" , "--region" , help = "AWS region to use, e.g. 'us-west-1'. " "Can use AWS_DEFAULT_REGION for local testing.  Use '" + LOCAL_REGION + "' for local DynamoDB testing" , ) parser . add_argument ( "--host" , help = "Host of local DynamoDB [required only for local]" ) parser . add_argument ( "--port" , help = "Port of local DynamoDB [required only for local]" ) parser . add_argument ( "--accessKey" , help = "Access key of local DynamoDB " "[required only for local]" ) parser . add_argument ( "--secretKey" , help = "Secret key of local DynamoDB " "[required only for local]" ) parser . add_argument ( "-p" , "--profile" , help = "AWS credentials file profile to use. Allows you to use a " "profile instead accessKey, secretKey authentication" , ) parser . add_argument ( "-s" , "--srcTable" , help = "Source DynamoDB table name to backup or restore from, " "use 'tablename*' for wildcard prefix selection or '*' for " "all tables.  Mutually exclusive with --tag" , ) parser . add_argument ( "-d" , "--destTable" , help = "Destination DynamoDB table name to backup or restore to, " "use 'tablename*' for wildcard prefix selection " "(defaults to use '-' separator) [optional, defaults to source]" , ) parser . add_argument ( "--prefixSeparator" , help = "Specify a different prefix separator, " "e.g. '.' [optional]" , ) parser . add_argument ( "--noSeparator" , action = "store_true" , help = "Overrides the use of a prefix separator for backup wildcard " "searches [optional]" , ) parser . add_argument ( "--readCapacity" , help = "Change the temp read capacity of the DynamoDB table to backup " "from [optional]" , ) parser . add_argument ( "-t" , "--tag" , help = "Tag to use for identifying tables to back up.  " "Mutually exclusive with srcTable.  Provided as KEY=VALUE" , ) parser . add_argument ( "--writeCapacity" , help = "Change the temp write capacity of the DynamoDB table to restore " "to [defaults to " + str ( RESTORE_WRITE_CAPACITY ) + ", optional]" , ) parser . add_argument ( "--schemaOnly" , action = "store_true" , default = False , help = "Backup or restore the schema only. Do not backup/restore data. " "Can be used with both backup and restore modes. Cannot be used with " "the --dataOnly [optional]" , ) parser . add_argument ( "--dataOnly" , action = "store_true" , default = False , help = "Restore data only. Do not delete/recreate schema [optional for " "restore]" , ) parser . add_argument ( "--noConfirm" , action = "store_true" , default = False , help = "Don't ask for confirmation before deleting existing schemas." , ) parser . add_argument ( "--skipThroughputUpdate" , action = "store_true" , default = False , help = "Skip updating throughput values across tables [optional]" , ) parser . add_argument ( "--dumpPath" , help = "Directory to place and search for DynamoDB table " "backups (defaults to use '" + str ( DATA_DUMP ) + "') [optional]" , default = str ( DATA_DUMP ) , ) parser . add_argument ( "--billingMode" , help = "Set billing mode between " + str ( PROVISIONED_BILLING_MODE ) + "|" + str ( PAY_PER_REQUEST_BILLING_MODE ) + " (defaults to use '" + str ( PROVISIONED_BILLING_MODE ) + "') [optional]" , choices = [ PROVISIONED_BILLING_MODE , PAY_PER_REQUEST_BILLING_MODE ] , default = str ( PROVISIONED_BILLING_MODE ) , ) parser . add_argument ( "--log" , help = "Logging level - DEBUG|INFO|WARNING|ERROR|CRITICAL " "[optional]" ) args = parser . parse_args ( ) # set log level log_level = LOG_LEVEL if args . log is not None : log_level = args . log . upper ( ) logging . basicConfig ( level = getattr ( logging , log_level ) ) # Check to make sure that --dataOnly and --schemaOnly weren't simultaneously specified if args . schemaOnly and args . dataOnly : logging . info ( "Options --schemaOnly and --dataOnly are mutually exclusive." ) sys . exit ( 1 ) # instantiate connection if args . region == LOCAL_REGION : conn = _get_aws_client ( service = "dynamodb" , access_key = args . accessKey , secret_key = args . secretKey , region = args . region , endpoint_url = "http://" + args . host + ":" + args . port , ) sleep_interval = LOCAL_SLEEP_INTERVAL else : if not args . profile : conn = _get_aws_client ( service = "dynamodb" , access_key = args . accessKey , secret_key = args . secretKey , region = args . region , ) sleep_interval = AWS_SLEEP_INTERVAL else : conn = _get_aws_client ( service = "dynamodb" , profile = args . profile , region = args . region , ) sleep_interval = AWS_SLEEP_INTERVAL # don't proceed if connection is not established if not conn : logging . info ( "Unable to establish connection with dynamodb" ) sys . exit ( 1 ) # set prefix separator prefix_separator = DEFAULT_PREFIX_SEPARATOR if args . prefixSeparator is not None : prefix_separator = args . prefixSeparator if args . noSeparator is True : prefix_separator = None # do backup/restore start_time = datetime . datetime . now ( ) . replace ( microsecond = 0 ) if args . mode == "backup" : matching_backup_tables = [ ] if args . tag : # Use Boto3 to find tags.  Boto3 provides a paginator that makes searching ta matching_backup_tables = get_table_name_by_tag ( args . profile , args . region , args . tag ) elif args . srcTable . find ( "*" ) != - 1 : matching_backup_tables = get_table_name_matches ( conn , args . srcTable , prefix_separator ) elif args . srcTable : matching_backup_tables . append ( args . srcTable ) if len ( matching_backup_tables ) == 0 : logging . info ( "No matching tables found.  Nothing to do." ) sys . exit ( 0 ) else : logging . info ( "Found " + str ( len ( matching_backup_tables ) ) + " table(s) in DynamoDB host to backup: " + ", " . join ( matching_backup_tables ) ) try : if args . srcTable . find ( "*" ) == - 1 : do_backup ( conn , args . read_capacity , tableQueue = None ) else : do_backup ( conn , args . read_capacity , matching_backup_tables ) except AttributeError : # Didn't specify srcTable if we get here q = Queue ( ) threads = [ ] for i in range ( MAX_NUMBER_BACKUP_WORKERS ) : t = threading . Thread ( target = do_backup , args = ( conn , args . readCapacity ) , kwargs = { "tableQueue" : q } , ) t . start ( ) threads . append ( t ) time . sleep ( THREAD_START_DELAY ) for table in matching_backup_tables : q . put ( table ) q . join ( ) for i in range ( MAX_NUMBER_BACKUP_WORKERS ) : q . put ( None ) for t in threads : t . join ( ) try : logging . info ( "Backup of table(s) " + args . srcTable + " completed!" ) except ( NameError , TypeError ) : logging . info ( "Backup of table(s) " + ", " . join ( matching_backup_tables ) + " completed!" ) if args . archive : if args . tag : for table in matching_backup_tables : dump_path = args . dumpPath + os . sep + table did_archive , archive_file = do_archive ( args . archive , dump_path ) if args . bucket and did_archive : do_put_bucket_object ( args . profile , args . region , args . bucket , archive_file ) else : did_archive , archive_file = do_archive ( args . archive , args . dumpPath ) if args . bucket and did_archive : do_put_bucket_object ( args . profile , args . region , args . bucket , archive_file ) elif args . mode == "restore" : if args . destTable is not None : dest_table = args . destTable else : dest_table = args . srcTable # If backups are in S3 download and extract the backup to use during restoration if args . bucket : do_get_s3_archive ( args . profile , args . region , args . bucket , args . srcTable , args . archive ) if dest_table . find ( "*" ) != - 1 : matching_destination_tables = get_table_name_matches ( conn , dest_table , prefix_separator ) delete_str = ": " if args . dataOnly else " to be deleted: " logging . info ( "Found " + str ( len ( matching_destination_tables ) ) + " table(s) in DynamoDB host" + delete_str + ", " . join ( matching_destination_tables ) ) threads = [ ] for table in matching_destination_tables : t = threading . Thread ( target = delete_table , args = ( conn , sleep_interval , table ) ) threads . append ( t ) t . start ( ) time . sleep ( THREAD_START_DELAY ) for thread in threads : thread . join ( ) matching_restore_tables = get_restore_table_matches ( args . srcTable , prefix_separator ) logging . info ( "Found " + str ( len ( matching_restore_tables ) ) + " table(s) in " + args . dumpPath + " to restore: " + ", " . join ( matching_restore_tables ) ) threads = [ ] for source_table in matching_restore_tables : if args . srcTable == "*" : t = threading . Thread ( target = do_restore , args = ( conn , sleep_interval , source_table , source_table , args . writeCapacity , args . billingMode , ) , ) else : t = threading . Thread ( target = do_restore , args = ( conn , sleep_interval , source_table , change_prefix ( source_table , args . srcTable , dest_table , prefix_separator , ) , args . writeCapacity , args . billingMode , ) , ) threads . append ( t ) t . start ( ) time . sleep ( THREAD_START_DELAY ) for thread in threads : thread . join ( ) logging . info ( "Restore of table(s) " + args . srcTable + " to " + dest_table + " completed!" ) else : delete_table ( conn = conn , sleep_interval = sleep_interval , table_name = dest_table ) do_restore ( dynamo = conn , sleep_interval = sleep_interval , source_table = args . srcTable , destination_table = dest_table , write_capacity = args . writeCapacity , billing_mode = args . billingMode , ) elif args . mode == "empty" : if args . srcTable . find ( "*" ) != - 1 : matching_backup_tables = get_table_name_matches ( conn , args . srcTable , prefix_separator ) logging . info ( "Found " + str ( len ( matching_backup_tables ) ) + " table(s) in DynamoDB host to empty: " + ", " . join ( matching_backup_tables ) ) threads = [ ] for table in matching_backup_tables : t = threading . Thread ( target = do_empty , args = ( conn , table , args . billingMode ) ) threads . append ( t ) t . start ( ) time . sleep ( THREAD_START_DELAY ) for thread in threads : thread . join ( ) logging . info ( "Empty of table(s) " + args . srcTable + " completed!" ) else : do_empty ( conn , args . srcTable , args . billingMode )