yage
7617c938b1
===================================================================== --- Changes ------------------------------ - python library functions moved to `py/lib` - python scwrypts renamed in kebob-case to help prevent import - __name__ == '__main__' enforced on all python scwrypts --- New Features ------------------------- - `__override` variables now allow values to be force-overwritten - py.lib.http.client provides a slim `requests.request` wrapper --- New Scripts -------------------------- py/data/convert ) quick data converters - csv-to-json - csv-to-yaml - json-to-csv - json-to-yaml - yaml-to-csv - yaml-to-json py/linear ) uses the linear.app graphql API for PM tasks - comment --- Bug Fixes ---------------------------- - `scwrypts` handles arguments with quotes and special characters
77 lines
2.2 KiB
Python
77 lines
2.2 KiB
Python
import csv
|
|
import json
|
|
import yaml
|
|
|
|
from py.lib.data.io import get_stream
|
|
|
|
|
|
def convert(input_file, input_type, output_file, output_type):
|
|
if input_type == output_type:
|
|
raise ValueError('input type and output type are the same')
|
|
|
|
with get_stream(input_file) as input_stream:
|
|
data = convert_input(input_stream, input_type)
|
|
|
|
with get_stream(output_file, 'w+') as output_stream:
|
|
_write_output(output_stream, output_type, data)
|
|
|
|
|
|
def convert_input(stream, input_type):
|
|
supported_input_types = {'csv', 'json', 'yaml'}
|
|
|
|
if input_type not in supported_input_types:
|
|
raise ValueError(f'input_type "{input_type}" not supported; must be one of {supported_input_types}')
|
|
|
|
return {
|
|
'csv': _read_csv,
|
|
'json': _read_json,
|
|
'yaml': _read_yaml,
|
|
}[input_type](stream)
|
|
|
|
def _write_output(stream, output_type, data):
|
|
supported_output_types = {'csv', 'json', 'yaml'}
|
|
|
|
if output_type not in supported_output_types:
|
|
raise ValueError(f'output_type "{output_type}" not supported; must be one of {supported_output_types}')
|
|
|
|
return {
|
|
'csv': _write_csv,
|
|
'json': _write_json,
|
|
'yaml': _write_yaml,
|
|
}[output_type](stream, data)
|
|
|
|
#####################################################################
|
|
|
|
def _read_csv(stream):
|
|
return [dict(line) for line in csv.DictReader(stream)]
|
|
|
|
def _write_csv(stream, data):
|
|
writer = csv.DictWriter(stream, fieldnames=list({
|
|
key
|
|
for dictionary in data
|
|
for key in dictionary.keys()
|
|
}))
|
|
|
|
writer.writeheader()
|
|
|
|
for value in data:
|
|
writer.writerow(value)
|
|
|
|
#####################################################################
|
|
|
|
def _read_json(stream):
|
|
data = json.loads(stream.read())
|
|
return data if isinstance(data, list) else [data]
|
|
|
|
def _write_json(stream, data):
|
|
stream.write(json.dumps(data))
|
|
|
|
#####################################################################
|
|
|
|
def _read_yaml(stream):
|
|
data = yaml.safe_load(stream)
|
|
return data if isinstance(data, list) else [data]
|
|
|
|
def _write_yaml(stream, data):
|
|
yaml.dump(data, stream, default_flow_style=False)
|