Python ruamel.yaml 模块，RoundTripDumper() 实例源码

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

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")