def save(self, vb=False):
if vb:
log.info("Saving...")
try:
with open('config.yaml', 'w') as fp:
yaml.dump(self.doc, fp, Dumper=yaml.RoundTripDumper)
except PermissionError:
log.err("No write access to config.yaml")
except IOError as e:
log.err("Could not open config.yaml: " + str(e))
except Exception as e:
log.err("An unexcpected exception of type: "
+ type(e).__name__
+ "has occurred: " + str(e))
else:
if vb:
log.info("Save complete")
return
python类RoundTripDumper()的实例源码
def yaml_save_roundtrip(filename, data, create_backup=False):
"""
Dump yaml using the RoundtripDumper and correct linespacing in output file
:param filename: name of the yaml file to save to
:param data: data structure to save
"""
if not EDITING_ENABLED:
return
sdata = yaml.dump(data, Dumper=yaml.RoundTripDumper, version=yaml_version, indent=indent_spaces, block_seq_indent=block_seq_indent, width=12288, allow_unicode=True)
# with open(filename+'_raw'+YAML_FILE, 'w') as outfile:
# outfile.write( sdata )
if create_backup:
if os.path.isfile(filename+YAML_FILE):
shutil.copy2(filename+YAML_FILE, filename+'.bak')
sdata = _format_yaml_dump2( sdata )
with open(filename+YAML_FILE, 'w') as outfile:
outfile.write( sdata )
def test_s3fetchyaml(mock_s3):
"""Success path all the way through to the mocked boto s3 object."""
input_dict = {'newkey': 'newvalue', 'newkey2': 'newvalue2'}
string_of_yaml = yaml.dump(input_dict, Dumper=yaml.RoundTripDumper)
bunch_of_bytes = bytes(string_of_yaml, 'utf-8')
mock_s3.side_effect = [{'Body': bunch_of_bytes}]
context = Context({
'k1': 'v1',
's3Fetch': {
'serviceName': 'service name',
'methodName': 'method_name',
'clientArgs': {'ck1': 'cv1', 'ck2': 'cv2'},
'methodArgs': {'Bucket': 'bucket name',
'Key': 'key name',
'SSECustomerAlgorithm': 'sse alg',
'SSECustomerKey': 'sse key'}
}})
pypyraws.steps.s3fetchyaml.run_step(context)
assert len(context) == 4
assert context['k1'] == 'v1'
assert context['newkey'] == 'newvalue'
assert context['newkey2'] == 'newvalue2'
def save_yaml(fname, wf, inline, pack, relpath, wd, encoding='utf-8'):
with codecs.open(fname, 'wb', encoding=encoding) as yaml_file:
yaml_file.write('#!/usr/bin/env cwl-runner\n')
yaml_file.write(yaml.dump(wf.to_obj(inline=inline,
pack=pack,
relpath=relpath,
wd=wd),
Dumper=yaml.RoundTripDumper))
def dump(self):
return ydump(self.data, Dumper=RoundTripDumper, default_flow_style=False)
def list():
"""
List all mod configuration
"""
config_path = get_default_config_path()
config = load_config(config_path, loader=yaml.RoundTripLoader)
print(yaml.dump(config['mod'], Dumper=yaml.RoundTripDumper))
def dump_config(config_path, config, dumper=yaml.RoundTripDumper):
with codecs.open(config_path, mode='w', encoding='utf-8') as file:
file.write(yaml.dump(config, Dumper=dumper))
def dump_config(config_path, config, dumper=yaml.RoundTripDumper):
with codecs.open(config_path, mode='w', encoding='utf-8') as stream:
stream.write(to_utf8(yaml.dump(config, Dumper=dumper)))
def write(self):
with open(self.filename+".tmp", 'w') as fid:
yaml.dump(self.data, fid, Dumper=yaml.RoundTripDumper)
move(self.filename+".tmp", self.filename)
def save_config():
def do_save_config():
with open(_CONFIG_FILE, 'w', encoding='utf-8') as file:
yaml.dump(config, file, Dumper=yaml.RoundTripDumper)
await get_event_loop().run_in_executor(None, do_save_config)
def task_config(args):
config_dict = _get_config_or_die(
calling_task='config',
required_params=[]
)
print(os.linesep.join((
'### yaml ###',
'',
yaml.dump(config_dict, Dumper=yaml.RoundTripDumper, indent=4),
'### /yaml ###'
)))
def create_user_says_skeleton(self):
template = os.path.join(self.template_dir, 'user_says.yaml')
skeleton = {}
for intent in self.assist._intent_action_funcs:
# print(type(intent))
entity_map_from_action = self.assist._intent_mappings.get(intent, {})
d = yaml.compat.ordereddict()
d['UserSays'] = [None, None]
d['Annotations'] = [None, None]
# d['Annotations'] = self.parse_annotations_from_action_mappings(intent)
data = yaml.comments.CommentedMap(d) # to preserve order w/o tags
skeleton[intent] = data
with open(template, 'a') as f:
f.write('# Template for defining UserSays examples\n\n')
f.write('# give-color-intent:\n\n')
f.write('# UserSays:\n')
f.write('# - My color is blue\n')
f.write('# - red is my favorite color\n\n')
f.write('# Annotations:\n')
f.write('# - blue: sys.color # maps param value -> entity\n')
f.write('# - red: sys.color\n\n\n\n')
# f.write(header)
yaml.dump(skeleton, f, default_flow_style=False, Dumper=yaml.RoundTripDumper)
def create_entity_skeleton(self):
print('Creating Template for Entities')
template = os.path.join(self.template_dir, 'entities.yaml')
message = """# Template file for entities\n\n"""
skeleton = {}
for intent in self.assist._intent_action_funcs:
entity_map = self.assist._intent_mappings.get(intent)
action_func = self.assist._intent_action_funcs[intent][0]
args = inspect.getargspec(action_func).args
# dont add API 'sys' entities to the template
if entity_map:
args = [a for a in args if 'sys.' not in entity_map.get(a, [])]
for param in [p for p in args if p not in skeleton]:
skeleton[param] = [None, None]
with open(template, 'w') as f:
f.write(message)
f.write('#Format as below\n\n')
f.write("# entity_name:\n")
f.write("# - entry1: list of synonyms \n")
f.write("# - entry2: list of synonyms \n\n")
f.write("#For example:\n\n")
f.write("# drink:\n")
f.write("# - water: ['aqua', 'h20'] \n")
f.write("# - coffee: ['joe', 'caffeine', 'espresso', 'late'] \n")
f.write("# - soda: ['pop', 'coke']\n\n\n\n")
yaml.dump(skeleton, f, default_flow_style=False, Dumper=yaml.RoundTripDumper)
def _yaml_save_roundtrip(filename, data):
"""
Dump yaml using the RoundtripDumper and correct linespacing in output file
"""
sdata = yaml.dump(data, Dumper=yaml.RoundTripDumper, version=yaml_version, indent=indent_spaces, block_seq_indent=2, width=12288, allow_unicode=True)
ldata = sdata.split('\n')
rdata = []
for index, line in enumerate(ldata):
# Fix for ruamel.yaml handling: Reinsert empty line before comment of next section
if len(line.lstrip()) > 0 and line.lstrip()[0] == '#':
indentcomment = len(line) - len(line.lstrip(' '))
indentprevline = len(ldata[index-1]) - len(ldata[index-1].lstrip(' '))
if indentprevline - indentcomment >= 2*indent_spaces:
rdata.append('')
rdata.append(line)
# Fix for ruamel.yaml handling: Remove empty line with spaces that have been inserted
elif line.strip() == '' and line != '':
if ldata[index-1] != '':
rdata.append(line)
else:
rdata.append(line)
sdata = '\n'.join(rdata)
if sdata[0] == '\n':
sdata =sdata[1:]
with open(filename+'.yaml', 'w') as outfile:
outfile.write( sdata )
def write_yaml(content, fpath):
yaml.dump(content, open(fpath, 'w'), Dumper=yaml.RoundTripDumper)
def dump_yaml(data, file_handle):
"""Wrapper function to nicely dump dictionaries as yaml files.
"""
assert(yaml is not None), "\nError: ruamel yaml python package not found."
yaml.dump(data, file_handle, Dumper=yaml.RoundTripDumper)
def start_line(self, document):
slicedpart = self._slice_segment(self._indices, document, include_selected=False)
if slicedpart is None or slicedpart == {} or slicedpart == []:
return 1
else:
return len(dump(slicedpart, Dumper=RoundTripDumper).rstrip().split('\n')) + 1
def end_line(self, document):
slicedpart = self._slice_segment(self._indices, document, include_selected=True)
return len(dump(slicedpart, Dumper=RoundTripDumper).rstrip().split('\n'))
def lines(self, document):
return "\n".join(dump(document, Dumper=RoundTripDumper).split('\n')[
self.start_line(document) - 1:self.end_line(document)
])
def lines_before(self, document, how_many):
return "\n".join(dump(document, Dumper=RoundTripDumper).split('\n')[
self.start_line(document) - 1 - how_many:self.start_line(document) - 1
])
def lines_after(self, document, how_many):
return "\n".join(dump(document, Dumper=RoundTripDumper).split('\n')[
self.end_line(document):self.end_line(document) + how_many
])
def run_step(context):
"""Parses input yaml file and substitutes {tokens} from context.
Loads yaml into memory to do parsing, so be aware of big files.
Args:
context: pypyr.context.Context. Mandatory.
The following context keys expected:
- fileFormatYamlIn. mandatory. path-like.
Path to source file on disk.
- fileFormatYamlOut. mandatory. path-like. Write output file to
here. Will create directories in path for you.
Returns:
None.
Raises:
FileNotFoundError: take a guess
pypyr.errors.KeyNotInContextError: fileFormatYamlIn or
fileFormatYamlOut missing in context.
pypyr.errors.KeyInContextHasNoValueError: fileFormatYamlIn or
fileFormatYamlOut exists but is None.
"""
logger.debug("started")
context.assert_keys_have_values(__name__,
'fileFormatYamlIn',
'fileFormatYamlOut')
in_path = context.get_formatted('fileFormatYamlIn')
out_path = context.get_formatted('fileFormatYamlOut')
logger.debug(f"opening yaml source file: {in_path}")
with open(in_path) as infile:
payload = yaml.load(infile, Loader=yaml.RoundTripLoader)
logger.debug(f"opening destination file for writing: {out_path}")
os.makedirs(os.path.abspath(os.path.dirname(out_path)), exist_ok=True)
with open(out_path, 'w') as outfile:
formatted_iterable = context.get_formatted_iterable(payload)
yaml.dump(formatted_iterable,
outfile,
Dumper=yaml.RoundTripDumper,
allow_unicode=True,
width=50)
logger.info(
f"Read {in_path} yaml, formatted contents and wrote to {out_path}")
logger.debug("done")
