Skip to content

Instantly share code, notes, and snippets.

Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save trueparallels/0415961b804c2f2bb5a645da00a42fc9 to your computer and use it in GitHub Desktop.
Save trueparallels/0415961b804c2f2bb5a645da00a42fc9 to your computer and use it in GitHub Desktop.
Convert a dynamodb result [json] to csv
#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "Elio Capelati Jr"
__version__ = "1.0.0"
import csv, json, sys, argparse
from pprint import pprint
def main(**kwargs):
values = transform_values(parse_json(kwargs['file_input']))
write_csv(values, kwargs['save'], kwargs['header'])
def transform_values(items):
"""
Transform a Item[n]:
{
"id": {
"N": "123"
},
"ttl": {
"N": "594777600"
},
...
}
Into:
{"id": "123", "ttl" :"594777600"}
"""
returnvalues = []
for item in items:
keys = item.keys()
itemloop = {}
for value in keys:
k, v = item.get(value).popitem()
itemloop[value] = v
returnvalues.append(itemloop)
return returnvalues
def parse_json(file_path):
data = json.load(args.input)
return data.get('Items', None)
def write_csv(values, save, header):
"""
Write a dict to CSV format including the header
"""
output = csv.writer(save)
if header:
output.writerow(values[0].keys())
for row in values:
output.writerow(row.values())
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='Parse DynamoDb table Items to a CSV')
parser.add_argument('input', nargs='?', type=argparse.FileType('r'),
default=sys.stdin,
help="""Input a Dynamodb JSON file ref:
http://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Programming.LowLevelAPI.html""")
parser.add_argument('output', nargs='?', type=argparse.FileType('w', encoding='UTF-8'),
default=sys.stdout,
help="""Provide a file name [csv], or a keep it blank and the content is sent to STDOUT""")
group = parser.add_mutually_exclusive_group()
group.add_argument('--csv-header', dest="header", action='store_true')
group.add_argument('--no-csv-header', dest="header", action='store_false')
parser.set_defaults(header=True)
args = parser.parse_args()
main(file_input=args.input,
save=args.output,
header=args.header)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment