yage
76a746a53e
=====================================================================
Notice the major version change which comes with breaking changes to
2.x! Reconstructs "library" functions for both python and zsh scwrypts,
with changes to virtualenv naming conventions (you'll need to refresh
all virtualenv with the appropriate scwrypt).
--- Changes ------------------------------
- changed a naming convention across zsh scripts, particularly
removing underscores where there is no need to avoid naming clash
(e.g. 'zsh/lib/utils/io.zsh' renames '__STATUS' to 'STATUS')
- moved clients reliant on py.lib.http to the py.lib.http module
- python scripts now rely on py.lib.scwrypts.execute
- updated package.json in zx scripts to include `type = module`
- 'scwrypts --list' commandline argument now includes additional
relevant data for each scwrypt
- environment variables no longer add themselves to be staged in the
'.env.template'
--- New Features -------------------------
- new 'use' syntax for disjoint import within zsh scripts; took me
a very long time to convince myself this would be necessary
- introduced scwrypt "groups" to allow portable module creation;
(i.e. ability add your own scripts from another repo!)
- py.lib.scwrypts.io provides a combined IO stream for quick, hybrid
use of input/output files and stdin/stdout
- py.lib.fzf provides a wrapper to provide similar functionality to
zsh/utils/io.zsh including fzf_(head|tail)
- improved efficiency of various scwrypts; notably reducing runtime
of scwrypts/environment sync
- improved scwrypts CLI by adding new options for exact scwrypt
matching, better filtering, and prettier/more-detailed interfaces
--- New Scripts --------------------------
- py/twilio )
basic SMS integration with twilio
- send-sms
- py/directus )
interactive directus GET query
- get-items
- py/discord )
post message to discord channel or webhook
- post-message
74 lines
2.0 KiB
Python
74 lines
2.0 KiB
Python
import csv
|
|
import json
|
|
import yaml
|
|
|
|
|
|
def convert(input_stream, input_type, output_stream, output_type):
|
|
if input_type == output_type:
|
|
raise ValueError('input type and output type are the same')
|
|
|
|
data = convert_input(input_stream, input_type)
|
|
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)
|