DynamoDBのダンプを取るプログラム。

dump.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
Get Dump from DynamoDB (All Tables).
This program does not cover all DynamoDB format,
but It's good enough for me.

restore.py can restore Tables from this outputs JSON files.

"""
import sys
import json
import time
import codecs
import decimal
import logging

import boto.dynamodb2

from boto.dynamodb2.fields import HashKey, RangeKey
from boto.dynamodb2.table import Table
from boto.dynamodb2.types import STRING, NUMBER

# --------------------------------
# Overwrite These Parameters
# --------------------------------

REGION='ap-southeast-1'
AWS_ACCESS_KEY_ID=''
AWS_SECRET_ACCESS_KEY=''

# --------------------------------

def find(pred, xs, else_value=None):
  for x in xs:
    if pred(x):
      return x
  return else_value


def typestr(v):
  ''' only recognize NUMBER and STRING '''
  nums_list = [int, float, decimal.Decimal]
  for t in nums_list:
    if isinstance(v, t):
      return NUMBER
  return STRING


def item_to_dict(item):
  res = {}
  for k in item.keys():
    res[k] = {
      'Value': '{}'.format(item[k]),
      'Type': typestr(item[k]),
    }
  return res


def _info_from(name, conn):
  table = Table(name, connection=conn)
  items = [ item_to_dict(item) for item in table.scan() ]
  desc = table.describe()['Table']
  
  hashkey = find(lambda x: x['KeyType'] == 'HASH', desc['KeySchema'])['AttributeName']
  result = {
    'TableName': name,
    'WriteCapacityUnits': desc['ProvisionedThroughput']['WriteCapacityUnits'],
    'ReadCapacityUnits': desc['ProvisionedThroughput']['ReadCapacityUnits'],
    'HashKey': {
      'Name': hashkey,
      'Type': find(lambda x: x['AttributeName'] == hashkey, desc['AttributeDefinitions'])['AttributeType'],
     },
     'Items': items,
  }
  
  r = find(lambda x: x['KeyType'] == 'RANGE', desc['KeySchema'])
  if r is None:
    return result
  rangekey = r['AttributeName']
  result['RangeKey'] = {
    'Name': rangekey,
    'Type': find(lambda x: x['AttributeName'] == rangekey, desc['AttributeDefinitions'])['AttributeType'],  
  }
  
  return result


def dump():
  conn = boto.dynamodb2.connect_to_region(
    REGION,
    aws_access_key_id=AWS_ACCESS_KEY_ID,
    aws_secret_access_key=AWS_SECRET_ACCESS_KEY
  )
  names = conn.list_tables()['TableNames']
  
  tables = map(lambda name: _info_from(name, conn), names)
  return json.dumps(tables)


if __name__ == '__main__':
  print dump()