Added schemaOnly and AWS credentials support (#23)

* Added schemaOnly and AWS credentials support

The --schemaOnly option allows the user to backup and restore
just the schema information for the given table(s). The use
case for this is if you need to migrate your tables from one
AWS account to a different AWS account.

(Our specific use case: We have an AWS account for development
and yet a separate AWS account for production.)

To support the use case of multiple AWS acount, -p or --profile
was also added so that the user could explicitly specify which
profile to use from the AWS credentials file (~/.aws/credentials).
This is /sightly/ more secure than putting accessKey and
secretKey into the command line.

* Merged with upstream

* BUGFIX: Indent was wrong on merged line

Corrected error. Was accidentally running a version in my path instead
of the one in CWD... Sorry.

Author: seliger

README.md

@@ -1,7 +1,7 @@
 dynamodump
 ==========
 
-[![Buildstatus](https://travis-ci.org/bchew/dynamodump.svg)](https://travis-ci.org/bchew/dynamodump)
+[![Buildstatus](https://travis-ci.org/bchew/dynamodump.svg)](https://travis-ci.org/bchew/dynamodump)		
 
 Simple backup and restore script for Amazon DynamoDB using boto to work similarly to mysqldump.
 
@@ -27,6 +27,9 @@ optional arguments:
   -r REGION, --region REGION
                         AWS region to use, e.g. 'us-west-1'. Use 'local' for
                         local DynamoDB testing.
+  -p PROFILE, --profile PROFILE
+                        AWS credentials file profile. Use as an alternative to
+                        putting accessKey and secretKey on the command line.
   -s SRCTABLE, --srcTable SRCTABLE
                         Source DynamoDB table name to backup or restore from,
                         use 'tablename*' for wildcard prefix selection or '*'
@@ -55,6 +58,9 @@ optional arguments:
                         Secret key of local DynamoDB [required only for local]
   --log LOG             Logging level - DEBUG|INFO|WARNING|ERROR|CRITICAL
                         [optional]
+  --schemaOnly          Dump or load schema only. Do not backup/restore data. 
+                        Can be used with both backup and restore modes. Cannot
+                        be used with the --dataOnly.
   --dataOnly            Restore data only. Do not delete/recreate schema
                         [optional for restore]
   --skipThroughputUpdate
@@ -93,6 +99,12 @@ python dynamodump.py -m backup -r us-west-1 -s "*"
 
 python dynamodump.py -m restore -r us-west-1 -s "*" --dataOnly
 ```
+Dump all table schemas and create the schemas (e.g. creating blank tables in a different AWS account):
+```
+python dynamodump.py -m backup -r us-west-1 -p source_credentials -s "*" --schemaOnly
+
+python dynamodump.py -m restore -r us-west-1 -p destination_credentials -s "*" --schemaOnly
+```
 
 Local example
 -------------

dynamodump.py

@@ -260,40 +260,41 @@ def do_backup(conn, table_name, read_capacity):
     f.write(json.dumps(table_desc, indent=JSON_INDENT))
     f.close()
 
-    original_read_capacity = table_desc["Table"]["ProvisionedThroughput"]["ReadCapacityUnits"]
-    original_write_capacity = table_desc["Table"]["ProvisionedThroughput"]["WriteCapacityUnits"]
+    if not args.schemaOnly:
+        original_read_capacity = table_desc["Table"]["ProvisionedThroughput"]["ReadCapacityUnits"]
+        original_write_capacity = table_desc["Table"]["ProvisionedThroughput"]["WriteCapacityUnits"]
 
-    # override table read capacity if specified
-    if read_capacity is not None and read_capacity != original_read_capacity:
-        update_provisioned_throughput(conn, table_name, read_capacity, original_write_capacity)
+        # override table read capacity if specified
+        if read_capacity is not None and read_capacity != original_read_capacity:
+            update_provisioned_throughput(conn, table_name, read_capacity, original_write_capacity)
 
-    # get table data
-    logging.info("Dumping table items for " + table_name)
-    mkdir_p(DUMP_PATH + "/" + table_name + "/" + DATA_DIR)
+        # get table data
+        logging.info("Dumping table items for " + table_name)
+        mkdir_p(DUMP_PATH + "/" + table_name + "/" + DATA_DIR)
 
-    i = 1
-    last_evaluated_key = None
+        i = 1
+        last_evaluated_key = None
 
-    while True:
-        scanned_table = conn.scan(table_name, exclusive_start_key=last_evaluated_key)
+        while True:
+            scanned_table = conn.scan(table_name, exclusive_start_key=last_evaluated_key)
 
-        f = open(DUMP_PATH + "/" + table_name + "/" + DATA_DIR + "/" + str(i).zfill(4) + ".json", "w+")
-        f.write(json.dumps(scanned_table, indent=JSON_INDENT))
-        f.close()
+            f = open(DUMP_PATH + "/" + table_name + "/" + DATA_DIR + "/" + str(i).zfill(4) + ".json", "w+")
+            f.write(json.dumps(scanned_table, indent=JSON_INDENT))
+            f.close()
 
-        i += 1
+            i += 1
 
-        try:
-            last_evaluated_key = scanned_table["LastEvaluatedKey"]
-        except KeyError:
-            break
+            try:
+                last_evaluated_key = scanned_table["LastEvaluatedKey"]
+            except KeyError:
+                break
 
-    # revert back to original table read capacity if specified
-    if read_capacity is not None and read_capacity != original_read_capacity:
-        update_provisioned_throughput(conn, table_name, original_read_capacity, original_write_capacity, False)
+        # revert back to original table read capacity if specified
+        if read_capacity is not None and read_capacity != original_read_capacity:
+            update_provisioned_throughput(conn, table_name, original_read_capacity, original_write_capacity, False)
 
-    logging.info("Backup for " + table_name + " table completed. Time taken: " + str(
-        datetime.datetime.now().replace(microsecond=0) - start_time))
+        logging.info("Backup for " + table_name + " table completed. Time taken: " + str(
+            datetime.datetime.now().replace(microsecond=0) - start_time))
 
 
 def do_restore(conn, sleep_interval, source_table, destination_table, write_capacity):
@@ -363,74 +364,78 @@ def do_restore(conn, sleep_interval, source_table, destination_table, write_capa
         # wait for table creation completion
         wait_for_active_table(conn, destination_table, "created")
 
-    # read data files
-    logging.info("Restoring data for " + destination_table + " table..")
-    data_file_list = os.listdir(dump_data_path + "/" + source_table + "/" + DATA_DIR + "/")
-    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 + "/" + source_table + "/" + DATA_DIR + "/" + 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 + "..")
+    if not args.schemaOnly:
+        # read data files
+        logging.info("Restoring data for " + destination_table + " table..")
+        data_file_list = os.listdir(dump_data_path + "/" + source_table + "/" + DATA_DIR + "/")
+        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 + "/" + source_table + "/" + DATA_DIR + "/" + 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(conn, sleep_interval, destination_table, put_requests)
+                    del put_requests[:]
+
+            # flush remainder
+            if len(put_requests) > 0:
                 batch_write(conn, sleep_interval, destination_table, put_requests)
-                del put_requests[:]
-
-        # flush remainder
-        if len(put_requests) > 0:
-            batch_write(conn, sleep_interval, destination_table, put_requests)
-
-    if not args.dataOnly and not args.skipThroughputUpdate:
-        # revert to original table write capacity if it has been modified
-        if write_capacity != original_write_capacity:
-            update_provisioned_throughput(conn, 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:
-                original_gsi_write_capacity = original_gsi_write_capacities.pop(0)
-                if original_gsi_write_capacity != gsi["ProvisionedThroughput"]["WriteCapacityUnits"]:
-                    gsi_data.append({"Update": {"IndexName": gsi["IndexName"],
-                                                "ProvisionedThroughput": {
-                                                    "ReadCapacityUnits": int(
-                                                        gsi["ProvisionedThroughput"]["ReadCapacityUnits"]),
-                                                    "WriteCapacityUnits": int(original_gsi_write_capacity)}}})
-
-            logging.info("Updating " + destination_table + " global secondary indexes write capacities as necessary..")
-            while True:
-                try:
-                    conn.update_table(destination_table, global_secondary_index_updates=gsi_data)
-                    break
-                except boto.exception.JSONResponseError as e:
-                    if e.body["__type"] == "com.amazonaws.dynamodb.v20120810#LimitExceededException":
-                        logging.info(
-                            "Limit exceeded, retrying updating throughput of GlobalSecondaryIndexes in " + destination_table + "..")
-                        time.sleep(sleep_interval)
-                    elif e.body["__type"] == "com.amazon.coral.availability#ThrottlingException":
-                        logging.info(
-                            "Control plane limit exceeded, retrying updating throughput of GlobalSecondaryIndexes in " + destination_table + "..")
-                        time.sleep(sleep_interval)
-
-    logging.info("Restore for " + source_table + " to " + destination_table + " table completed. Time taken: " + str(
-        datetime.datetime.now().replace(microsecond=0) - start_time))
 
+        if not args.dataOnly and not args.skipThroughputUpdate:
+            # revert to original table write capacity if it has been modified
+            if write_capacity != original_write_capacity:
+                update_provisioned_throughput(conn, 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:
+                    original_gsi_write_capacity = original_gsi_write_capacities.pop(0)
+                    if original_gsi_write_capacity != gsi["ProvisionedThroughput"]["WriteCapacityUnits"]:
+                        gsi_data.append({"Update": {"IndexName": gsi["IndexName"],
+                                                    "ProvisionedThroughput": {
+                                                        "ReadCapacityUnits": int(
+                                                            gsi["ProvisionedThroughput"]["ReadCapacityUnits"]),
+                                                        "WriteCapacityUnits": int(original_gsi_write_capacity)}}})
+
+                logging.info("Updating " + destination_table + " global secondary indexes write capacities as necessary..")
+                while True:
+                    try:
+                        conn.update_table(destination_table, global_secondary_index_updates=gsi_data)
+                        break
+                    except boto.exception.JSONResponseError as e:
+                        if e.body["__type"] == "com.amazonaws.dynamodb.v20120810#LimitExceededException":
+                            logging.info(
+                                "Limit exceeded, retrying updating throughput of GlobalSecondaryIndexes in " + destination_table + "..")
+                            time.sleep(sleep_interval)
+                        elif e.body["__type"] == "com.amazon.coral.availability#ThrottlingException":
+                            logging.info(
+                                "Control plane limit exceeded, retrying updating throughput of GlobalSecondaryIndexes in " + destination_table + "..")
+                            time.sleep(sleep_interval)
+
+        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))
 
 # parse args
 parser = argparse.ArgumentParser(description="Simple DynamoDB backup/restore/empty.")
 parser.add_argument("-m", "--mode", help="'backup' or 'restore' or 'empty'")
 parser.add_argument("-r", "--region",
                     help="AWS region to use, e.g. 'us-west-1'. Use '" + LOCAL_REGION + "' for local DynamoDB testing.")
+parser.add_argument("-p", "--profile",
+                    help="AWS profile name to use. Allows to you to choose alternate profile from your AWS credentials file.")
 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.")
 parser.add_argument("-d", "--destTable",
@@ -448,6 +453,8 @@ def do_restore(conn, sleep_interval, source_table, destination_table, write_capa
 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("--log", help="Logging level - DEBUG|INFO|WARNING|ERROR|CRITICAL [optional]")
+parser.add_argument("--schemaOnly", action="store_true", default=False,
+                    help="Backup or restore the schema only. Do not export/import data. [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("--skipThroughputUpdate", action="store_true", default=False,
@@ -460,15 +467,27 @@ def do_restore(conn, sleep_interval, source_table, destination_table, write_capa
     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 = DynamoDBConnection(aws_access_key_id=args.accessKey, aws_secret_access_key=args.secretKey, host=args.host,
                               port=int(args.port), is_secure=False)
     sleep_interval = LOCAL_SLEEP_INTERVAL
 else:
-    conn = boto.dynamodb2.connect_to_region(args.region, aws_access_key_id=args.accessKey,
-                                            aws_secret_access_key=args.secretKey)
-    sleep_interval = AWS_SLEEP_INTERVAL
+    if not args.profile:
+        conn = boto.dynamodb2.connect_to_region(args.region, aws_access_key_id=args.accessKey,
+                                                aws_secret_access_key=args.secretKey)
+        sleep_interval = AWS_SLEEP_INTERVAL
+    else:
+        conn = boto.dynamodb2.connect_to_region(args.region, profile_name=args.profile)
+        sleep_interval = AWS_SLEEP_INTERVAL
+
 
 # set prefix separator
 prefix_separator = DEFAULT_PREFIX_SEPARATOR