Mercurial > hg > Lemuriformes

comparison lemuriformes/serialize.py @ 17:4793f99b73e0

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
[lemuriformes] utility functions
author Jeff Hammel <k0scist@gmail.com>
date Sun, 10 Dec 2017 17:42:52 -0800
parents
children
comparison
equal deleted inserted replaced
16:9b1bb9eee962 17:4793f99b73e0
1 #!/usr/bin/env python
2
3 """
4 serialization
5 """
6
7 import argparse
8 import csv
9 import json
10 import sys
11 from StringIO import StringIO
12 from .cast import isstring
13 from .cast import unify
14
15
16 def dictlist2csv(list_of_dicts, header=None, fp=None):
17 """
18 convert a `list_of_dicts` to CSV
19
20 `fp` should be a file-like object or a path to a file
21 that will be overwritten. If `fp` is not provided, a
22 string will be returned
23 """
24
25 if isstring(fp):
26 with open(fp, 'w') as _fp:
27 return dictlist2csv(list_of_dicts, _fp)
28
29 return_string = False
30 if fp is None:
31 return_string = True
32 fp = StringIO()
33
34 # get the header
35 if not list_of_dicts:
36 return # XXX what about return_string? Good or bad?
37 header = header or list_of_dicts[0].keys()
38
39 # instantiate a writer
40 writer = csv.DictWriter(fp, fieldnames=header)
41 writer.writeheader()
42 for row in list_of_dicts:
43 row = {key: unify(value)
44 for key, value in row.items()}
45 try:
46 writer.writerow(row)
47 except (UnicodeDecodeError, UnicodeEncodeError) as e:
48 print (row )
49 print(e)
50 raise
51
52 if return_string:
53 return fp.getvalue()
54
55
56 def dump_json(_json):
57 """general purpose JSON front-end"""
58 return json.dumps(_json, indent=2, sort_keys=True)
59
60
61 def append(filename, item):
62 """append line-`item` to `filename`"""
63
64 with open(filename, 'a') as f:
65 f.write('{}\n'.format(item))
66
67
68 def main(args=sys.argv[1:]):
69 """CLI"""
70
71 # parse command line
72 description = "convert a list of dicts in JSON format to CSV"
73 parser = argparse.ArgumentParser(description=description)
74 parser.add_argument('input',
75 type=argparse.FileType('r'),
76 help="path to file containing a list of flat dicts")
77 parser.add_argument('-o', '--output', dest='output',
78 type=argparse.FileType('w'), default=sys.stdout,
79 help="file to write the CSV to [DEFAULT: stdout]")
80 options = parser.parse_args(args)
81
82 # parse input
83 data = json.loads(options.input.read())
84 assert type(data) == list
85
86 # write output
87 dictlist2csv(data, options.output)
88
89
90 if __name__ == '__main__':
91 main()