text: #!/usr/bin/env python3
  """
      Source: https://github.com/bchew/dynamodump
      License:
  
      Copyright 2013 Benny Chew
  
      Permission is hereby granted, free of charge, to any person obtaining a copy of
      this software and associated documentation files (the "Software"), to deal in
      the Software without restriction, including without limitation the rights to
      use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
      of the Software, and to permit persons to whom the Software is furnished to do
      so, subject to the following conditions:
  
      The above copyright notice and this permission notice shall be included in all
      copies or substantial portions of the Software.
  
      THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
      IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
      FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
      AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
      LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
      OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
      SOFTWARE.
  
      Simple backup and restore script for Amazon DynamoDB using boto to work similarly to mysqldump.
  
      Suitable for DynamoDB usages of smaller data volume which do not warrant the usage of AWS
      Data Pipeline for backup/restores/empty.
  
      dynamodump supports local DynamoDB instances as well (tested with dynalite).
  """
  
  import argparse
  import fnmatch
  import json
  import logging
  import os
  import shutil
  import threading
  import datetime
  import errno
  import sys
  import time
  import re
  import zipfile
  import tarfile
  
  try:
      from queue import Queue
  except ImportError:
      from Queue import Queue
  
  try:
      from urllib.request import urlopen
      from urllib.error import URLError, HTTPError
  except ImportError:
      from urllib2 import urlopen, URLError, HTTPError
  
  import boto.dynamodb2.layer1
  from boto.dynamodb2.exceptions import ProvisionedThroughputExceededException
  import botocore
  import boto3
  
  
  JSON_INDENT = 2
  AWS_SLEEP_INTERVAL = 10  # seconds
  LOCAL_SLEEP_INTERVAL = 1  # seconds
  BATCH_WRITE_SLEEP_INTERVAL = 0.15  # seconds
  MAX_BATCH_WRITE = 25  # DynamoDB limit
  SCHEMA_FILE = "schema.json"
  DATA_DIR = "data"
  MAX_RETRY = 6
  LOCAL_REGION = "local"
  LOG_LEVEL = "INFO"
  DATA_DUMP = "dump"
  RESTORE_WRITE_CAPACITY = 25
  THREAD_START_DELAY = 1  # seconds
  CURRENT_WORKING_DIR = os.getcwd()
  DEFAULT_PREFIX_SEPARATOR = "-"
  MAX_NUMBER_BACKUP_WORKERS = 25
  METADATA_URL = "http://169.254.169.254/latest/meta-data/"
  
  
  def _get_aws_client(profile, region, service):
      """
      Build connection to some AWS service.
      """
  
      if region:
          aws_region = region
      else:
          aws_region = os.getenv("AWS_DEFAULT_REGION")
  
      # Fallback to querying metadata for region
      if not aws_region:
          try:
              azone = urlopen(METADATA_URL + "placement/availability-zone",
                              data=None, timeout=5).read().decode()
              aws_region = azone[:-1]
          except URLError:
              logging.exception("Timed out connecting to metadata service.\n\n")
              sys.exit(1)
          except HTTPError as e:
              logging.exception("Error determining region used for AWS client.  Typo in code?\n\n" +
                                str(e))
              sys.exit(1)
  
      if profile:
          session = boto3.Session(profile_name=profile)
          client = session.client(service, region_name=aws_region)
      else:
          client = boto3.client(service, region_name=aws_region)
      return client
  
  
  def get_table_name_by_tag(profile, region, tag):
      """
      Using provided connection to dynamodb and tag, get all tables that have provided tag
  
      Profile provided and, if needed, used to build connection to STS.
      """
  
      matching_tables = []
      all_tables = []
      sts = _get_aws_client(profile, region, "sts")
      dynamo = _get_aws_client(profile, region, "dynamodb")
      account_number = sts.get_caller_identity().get("Account")
      paginator = dynamo.get_paginator("list_tables")
      tag_key = tag.split("=")[0]
      tag_value = tag.split("=")[1]
  
      get_all_tables = paginator.paginate()
      for page in get_all_tables:
          for table in page["TableNames"]:
              all_tables.append(table)
              logging.debug("Found table " + table)
  
      for table in all_tables:
          table_arn = "arn:aws:dynamodb:{}:{}:table/{}".format(
              region, account_number, table)
          table_tags = dynamo.list_tags_of_resource(
              ResourceArn=table_arn
          )
          for found_tag in table_tags["Tags"]:
              if found_tag["Key"] == tag_key:
                  logging.debug("Checking table " + table +
                                " tag " + found_tag["Key"])
                  if found_tag["Value"] == tag_value:
                      matching_tables.append(table)
                      logging.info("Matched table " + table)
  
      return matching_tables
  
  
  def do_put_bucket_object(profile, region, bucket, bucket_object):
      """
      Put object into bucket.  Only called if we've also created an archive file with do_archive()
  
      Bucket must exist prior to running this function.
      profile could be None.
      bucket_object is file to be uploaded
      """
  
      s3 = _get_aws_client(profile, region, "s3")
      logging.info("Uploading backup to S3 bucket " + bucket)
      try:
          s3.upload_file(bucket_object, bucket, bucket_object,
                         ExtraArgs={
                             "ServerSideEncryption": "AES256"
                         })
      except botocore.exceptions.ClientError as e:
          logging.exception("Failed to put file to S3 bucket\n\n" + str(e))
          sys.exit(1)
  
  
  def do_get_s3_archive(profile, region, bucket, table, archive):
      """
      Fetch latest file named filename from S3
  
      Bucket must exist prior to running this function.
      filename is args.dumpPath.  File would be "args.dumpPath" with suffix .tar.bz2 or .zip
      """
  
      s3 = _get_aws_client(profile, region, "s3")
  
      if archive:
          if archive == "tar":
              archive_type = "tar.bz2"
          else:
              archive_type = "zip"
  
      # Make sure bucket exists before continuing
      try:
          s3.head_bucket(
              Bucket=bucket
          )
      except botocore.exceptions.ClientError as e:
          logging.exception("S3 bucket " + bucket + " does not exist. "
                            "Can't get backup file\n\n" + str(e))
          sys.exit(1)
  
      try:
          contents = s3.list_objects_v2(
              Bucket=bucket,
              Prefix=args.dumpPath
          )
      except botocore.exceptions.ClientError as e:
          logging.exception(
              "Issue listing contents of bucket " + bucket + "\n\n" + str(e))
          sys.exit(1)
  
      # Script will always overwrite older backup.  Bucket versioning stores multiple backups.
      # Therefore, just get item from bucket based on table name since that's what we name the files.
      filename = None
      for d in contents["Contents"]:
          if d["Key"] == "{}/{}.{}".format(args.dumpPath, table, archive_type):
              filename = d["Key"]
  
      if not filename:
          logging.exception("Unable to find file to restore from.  "
                            "Confirm the name of the table you're restoring.")
          sys.exit(1)
  
      output_file = "/tmp/" + os.path.basename(filename)
      logging.info("Downloading file " + filename + " to " + output_file)
      s3.download_file(bucket, filename, output_file)
  
      # Extract archive based on suffix
      if tarfile.is_tarfile(output_file):
          try:
              logging.info("Extracting tar file...")
              with tarfile.open(name=output_file, mode="r:bz2") as a:
                  a.extractall(path=".")
          except tarfile.ReadError as e:
              logging.exception("Error reading downloaded archive\n\n" + str(e))
              sys.exit(1)
          except tarfile.ExtractError as e:
              # ExtractError is raised for non-fatal errors on extract method
              logging.error("Error during extraction: " + str(e))
  
      # Assuming zip file here since we're only supporting tar and zip at this time
      else:
          try:
              logging.info("Extracting zip file...")
              with zipfile.ZipFile(output_file, "r") as z:
                  z.extractall(path=".")
          except zipfile.BadZipFile as e:
              logging.exception("Problem extracting zip file\n\n" + str(e))
              sys.exit(1)
  
  
  def do_archive(archive_type, dump_path):
      """
      Create compressed archive of dump_path.
  
      Accepts archive_type of zip or tar and requires dump_path, directory added to archive
      """
  
      archive_base = dump_path
  
      if archive_type.lower() == "tar":
          archive = archive_base + ".tar.bz2"
          try:
              logging.info("Creating tar file " + archive + "...")
              with tarfile.open(name=archive, mode="w:bz2") as a:
                  for root, dirs, files in os.walk(archive_base):
                      for file in files:
                          a.add(os.path.join(root, file))
                  return True, archive
          except tarfile.CompressionError as e:
              logging.exception("compression method is not supported or the data cannot be"
                                " decoded properly.\n\n" + str(e))
              sys.exit(1)
          except tarfile.TarError as e:
              logging.exception("Error creating tarfile archive.\n\n" + str(e))
              sys.exit(1)
  
      elif archive_type.lower() == "zip":
          try:
              logging.info("Creating zip file...")
              archive = archive_base + ".zip"
              with zipfile.ZipFile(archive, "w") as z:
                  for root, dirs, files in os.walk(archive_base):
                      for file in files:
                          z.write(os.path.join(root, file))
                  return True, archive
          except zipfile.BadZipFile as e:
              logging.exception("Problem creating zip file\n\n" + str(e))
              sys.exit(1)
          except zipfile.LargeZipFile:
              logging.exception(
                  "Zip file would be too large.  Update code to use Zip64 to continue.")
              sys.exit(1)
  
      else:
          logging.error("Unsupported archive format received.  Probably shouldn't have "
                        "made it to this code path.  Skipping attempt at creating archive file")
          return False, None
  
  
  def get_table_name_matches(conn, table_name_wildcard, separator):
      """
      Find tables to backup
      """
  
      all_tables = []
      last_evaluated_table_name = None
  
      while True:
          table_list = conn.list_tables(
              exclusive_start_table_name=last_evaluated_table_name)
          all_tables.extend(table_list["TableNames"])
  
          try:
              last_evaluated_table_name = table_list["LastEvaluatedTableName"]
          except KeyError:
              break
  
      matching_tables = []
      for table_name in all_tables:
          if fnmatch.fnmatch(table_name, table_name_wildcard):
              logging.info("Adding %s", table_name)
              matching_tables.append(table_name)
  
      return matching_tables
  
  
  def get_restore_table_matches(table_name_wildcard, separator):
      """
      Find tables to restore
      """
  
      matching_tables = []
      try:
          dir_list = os.listdir("./" + args.dumpPath)
      except OSError:
          logging.info("Cannot find \"./%s\", Now trying current working directory.."
                       % args.dumpPath)
          dump_data_path = CURRENT_WORKING_DIR
          try:
              dir_list = os.listdir(dump_data_path)
          except OSError:
              logging.info("Cannot find \"%s\" directory containing dump files!"
                           % dump_data_path)
              sys.exit(1)
  
      for dir_name in dir_list:
          if table_name_wildcard == "*":
              matching_tables.append(dir_name)
          elif separator == "":
  
              if dir_name.startswith(re.sub(r"([A-Z])", r" \1", table_name_wildcard.split("*", 1)[0])
                                     .split()[0]):
                  matching_tables.append(dir_name)
          elif dir_name.split(separator, 1)[0] == table_name_wildcard.split("*", 1)[0]:
              matching_tables.append(dir_name)
  
      return matching_tables
  
  
  def change_prefix(source_table_name, source_wildcard, destination_wildcard, separator):
      """
      Update prefix used for searching tables
      """
  
      source_prefix = source_wildcard.split("*", 1)[0]
      destination_prefix = destination_wildcard.split("*", 1)[0]
      if separator == "":
          if re.sub(r"([A-Z])", r" \1", source_table_name).split()[0] == source_prefix:
              return destination_prefix + re.sub(r"([A-Z])", r" \1", source_table_name)\
                  .split(" ", 1)[1].replace(" ", "")
      if source_table_name.split(separator, 1)[0] == source_prefix:
          return destination_prefix + separator + source_table_name.split(separator, 1)[1]
  
  
  def delete_table(conn, sleep_interval, table_name):
      """
      Delete table table_name
      """
  
      if not args.dataOnly:
          while True:
              # delete table if exists
              table_exist = True
              try:
                  conn.delete_table(table_name)
              except boto.exception.JSONResponseError as e:
                  if e.body["__type"] == "com.amazonaws.dynamodb.v20120810#ResourceNotFoundException":
                      table_exist = False
                      logging.info(table_name + " table deleted!")
                      break
                  elif e.body["__type"] == "com.amazonaws.dynamodb.v20120810#LimitExceededException":
                      logging.info(
                          "Limit exceeded, retrying deletion of " + table_name + "..")
                      time.sleep(sleep_interval)
                  elif e.body["__type"] == "com.amazon.coral.availability#ThrottlingException":
                      logging.info("Control plane limit exceeded, retrying deletion of " +
                                   table_name + "..")
                      time.sleep(sleep_interval)
                  elif e.body["__type"] == "com.amazonaws.dynamodb.v20120810#ResourceInUseException":
                      logging.info(table_name + " table is being deleted..")
                      time.sleep(sleep_interval)
                  else:
                      logging.exception(e)
                      sys.exit(1)
  
          # if table exists, wait till deleted
          if table_exist:
              try:
                  while True:
                      logging.info("Waiting for " + table_name + " table to be deleted.. [" +
                                   conn.describe_table(table_name)["Table"]["TableStatus"] + "]")
                      time.sleep(sleep_interval)
              except boto.exception.JSONResponseError as e:
                  if e.body["__type"] == "com.amazonaws.dynamodb.v20120810#ResourceNotFoundException":
                      logging.info(table_name + " table deleted.")
                      pass
                  else:
                      logging.exception(e)
                      sys.exit(1)
  
  
  def mkdir_p(path):
      """
      Create directory to hold dump
      """
  
      try:
          os.makedirs(path)
      except OSError as exc:
          if exc.errno == errno.EEXIST and os.path.isdir(path):
              pass
          else:
              raise
  
  
  def batch_write(conn, sleep_interval, table_name, put_requests):
      """
      Write data to table_name
      """
  
      request_items = {table_name: put_requests}
      i = 1
      sleep = sleep_interval
      while True:
          response = conn.batch_write_item(request_items)
          unprocessed_items = response["UnprocessedItems"]
  
          if len(unprocessed_items) == 0:
              break
          if len(unprocessed_items) > 0 and i <= MAX_RETRY:
              logging.debug(str(len(unprocessed_items)) +
                            " unprocessed items, retrying after %s seconds.. [%s/%s]"
                            % (str(sleep), str(i), str(MAX_RETRY)))
              request_items = unprocessed_items
              time.sleep(sleep)
              sleep += sleep_interval
              i += 1
          else:
              logging.info("Max retries reached, failed to processed batch write: " +
                           json.dumps(unprocessed_items, indent=JSON_INDENT))
              logging.info("Ignoring and continuing..")
              break
  
  
  def wait_for_active_table(conn, table_name, verb):
      """
      Wait for table to be indesired state
      """
  
      while True:
          if conn.describe_table(table_name)["Table"]["TableStatus"] != "ACTIVE":
              logging.info("Waiting for " + table_name + " table to be " + verb + ".. [" +
                           conn.describe_table(table_name)["Table"]["TableStatus"] + "]")
              time.sleep(sleep_interval)
          else:
              logging.info(table_name + " " + verb + ".")
              break
  
  
  def update_provisioned_throughput(conn, table_name, read_capacity, write_capacity, wait=True):
      """
      Update provisioned throughput on the table to provided values
      """
  
      logging.info("Updating " + table_name + " table read capacity to: " +
                   str(read_capacity) + ", write capacity to: " + str(write_capacity))
      while True:
          try:
              conn.update_table(table_name,
                                {"ReadCapacityUnits": int(read_capacity),
                                 "WriteCapacityUnits": int(write_capacity)})
              break
          except boto.exception.JSONResponseError as e:
              if e.body["__type"] == "com.amazonaws.dynamodb.v20120810#LimitExceededException":
                  logging.info(
                      "Limit exceeded, retrying updating throughput of " + table_name + "..")
                  time.sleep(sleep_interval)
              elif e.body["__type"] == "com.amazon.coral.availability#ThrottlingException":
                  logging.info("Control plane limit exceeded, retrying updating throughput"
                               "of " + table_name + "..")
                  time.sleep(sleep_interval)
  
      # wait for provisioned throughput update completion
      if wait:
          wait_for_active_table(conn, table_name, "updated")
  
  
  def do_empty(dynamo, table_name):
      """
      Empty table named table_name
      """
  
      logging.info("Starting Empty for " + table_name + "..")
  
      # get table schema
      logging.info("Fetching table schema for " + table_name)
      table_data = dynamo.describe_table(table_name)
  
      table_desc = table_data["Table"]
      table_attribute_definitions = table_desc["AttributeDefinitions"]
      table_key_schema = table_desc["KeySchema"]
      original_read_capacity = table_desc["ProvisionedThroughput"]["ReadCapacityUnits"]
      original_write_capacity = table_desc["ProvisionedThroughput"]["WriteCapacityUnits"]
      table_local_secondary_indexes = table_desc.get("LocalSecondaryIndexes")
      table_global_secondary_indexes = table_desc.get("GlobalSecondaryIndexes")
  
      table_provisioned_throughput = {"ReadCapacityUnits": int(original_read_capacity),
                                      "WriteCapacityUnits": int(original_write_capacity)}
  
      logging.info("Deleting Table " + table_name)
  
      delete_table(dynamo, sleep_interval, table_name)
  
      logging.info("Creating Table " + table_name)
  
      while True:
          try:
              dynamo.create_table(table_attribute_definitions, table_name, table_key_schema,
                                  table_provisioned_throughput, table_local_secondary_indexes,
                                  table_global_secondary_indexes)
              break
          except boto.exception.JSONResponseError as e:
              if e.body["__type"] == "com.amazonaws.dynamodb.v20120810#LimitExceededException":
                  logging.info(
                      "Limit exceeded, retrying creation of " + table_name + "..")
                  time.sleep(sleep_interval)
              elif e.body["__type"] == "com.amazon.coral.availability#ThrottlingException":
                  logging.info("Control plane limit exceeded, retrying creation of " +
                               table_name + "..")
                  time.sleep(sleep_interval)
              else:
                  logging.exception(e)
                  sys.exit(1)
  
      # wait for table creation completion
      wait_for_active_table(dynamo, table_name, "created")
  
      logging.info("Recreation of " + table_name + " completed. Time taken: " + str(
          datetime.datetime.now().replace(microsecond=0) - start_time))
  
  
  def do_backup(dynamo, read_capacity, tableQueue=None, srcTable=None):
      """
      Connect to DynamoDB and perform the backup for srcTable or each table in tableQueue
      """
  
      if srcTable:
          table_name = srcTable
  
      if tableQueue:
          while True:
              table_name = tableQueue.get()
              if table_name is None:
                  break
  
              logging.info("Starting backup for " + table_name + "..")
  
              # trash data, re-create subdir
              if os.path.exists(args.dumpPath + os.sep + table_name):
                  shutil.rmtree(args.dumpPath + os.sep + table_name)
              mkdir_p(args.dumpPath + os.sep + table_name)
  
              # get table schema
              logging.info("Dumping table schema for " + table_name)
              f = open(args.dumpPath + os.sep + table_name +
                       os.sep + SCHEMA_FILE, "w+")
              table_desc = dynamo.describe_table(table_name)
              f.write(json.dumps(table_desc, indent=JSON_INDENT))
              f.close()
  
              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(dynamo, table_name,
                                                    read_capacity, original_write_capacity)
  
                  # get table data
                  logging.info("Dumping table items for " + table_name)
                  mkdir_p(args.dumpPath + os.sep +
                          table_name + os.sep + DATA_DIR)
  
                  i = 1
                  last_evaluated_key = None
  
                  while True:
                      try:
                          scanned_table = dynamo.scan(table_name,
                                                      exclusive_start_key=last_evaluated_key)
                      except ProvisionedThroughputExceededException:
                          logging.error("EXCEEDED THROUGHPUT ON TABLE " +
                                        table_name + ".  BACKUP FOR IT IS USELESS.")
                          tableQueue.task_done()
  
                      f = open(
                          args.dumpPath + os.sep + table_name + os.sep + DATA_DIR + os.sep +
                          str(i).zfill(4) + ".json", "w+"
                      )
                      f.write(json.dumps(scanned_table, indent=JSON_INDENT))
                      f.close()
  
                      i += 1
  
                      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(dynamo,
                                                    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))
  
              tableQueue.task_done()
  
  
  def do_restore(dynamo, sleep_interval, source_table, destination_table, write_capacity):
      """
      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
  
      # override GSI write capacities if specified, else use RESTORE_WRITE_CAPACITY if original
      # write capacity is lower
      original_gsi_write_capacities = []
      if table_global_secondary_indexes is not None:
          for gsi in table_global_secondary_indexes:
              original_gsi_write_capacities.append(
                  gsi["ProvisionedThroughput"]["WriteCapacityUnits"])
  
              if gsi["ProvisionedThroughput"]["WriteCapacityUnits"] < int(write_capacity):
                  gsi["ProvisionedThroughput"]["WriteCapacityUnits"] = int(
                      write_capacity)
  
      # temp provisioned throughput for restore
      table_provisioned_throughput = {"ReadCapacityUnits": int(original_read_capacity),
                                      "WriteCapacityUnits": int(write_capacity)}
  
      if not args.dataOnly:
  
          logging.info("Creating " + destination_table + " table with temp write capacity of " +
                       str(write_capacity))
  
          while True:
              try:
                  dynamo.create_table(table_attribute_definitions, table_table_name, table_key_schema,
                                      table_provisioned_throughput, table_local_secondary_indexes,
                                      table_global_secondary_indexes)
                  break
              except boto.exception.JSONResponseError as e:
                  if e.body["__type"] == "com.amazonaws.dynamodb.v20120810#LimitExceededException":
                      logging.info(
                          "Limit exceeded, retrying creation of " + destination_table + "..")
                      time.sleep(sleep_interval)
                  elif e.body["__type"] == "com.amazon.coral.availability#ThrottlingException":
                      logging.info("Control plane limit exceeded, "
                                   "retrying creation of " + destination_table + "..")
                      time.sleep(sleep_interval)
                  else:
                      logging.exception(e)
                      sys.exit(1)
  
          # wait for table creation completion
          wait_for_active_table(dynamo, destination_table, "created")
      else:
          # 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:
                  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)
                      if original_gsi_write_capacity != wcu:
                          gsi_data.append({
                              "Update": {
                                  "IndexName": gsi["IndexName"],
                                  "ProvisionedThroughput": {
                                      "ReadCapacityUnits":
                                          int(rcu),
                                      "WriteCapacityUnits": int(original_gsi_write_capacity)
                                  }
                              }
                          })
  
                  logging.info("Updating " + destination_table +
                               " global secondary indexes write capacities as necessary..")
                  while True:
                      try:
                          dynamo.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)
  
          # 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 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("--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("--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 = boto.dynamodb2.layer1.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:
          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
  
      # 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))
                  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))
                  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, sleep_interval, dest_table)
              do_restore(conn, sleep_interval, args.srcTable,
                         dest_table, args.writeCapacity)
      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))
                  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)
  
  
  if __name__ == "__main__":
      main()
  