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

def import_config(args, input_file=None):
    if not input_file:
        input_file = sys.stdin
    source = input_file.read().strip()
    if source[0] == '{':
        # JSON input
        config = json.loads(source)
    else:
        # YAML input
        config = yaml.round_trip_load(source)

    STATE['stages'] = config['stages']
    config['config'] = _encrypt_dict(config['config'])
    with open(args.config, 'wt') as f:
        if config:
            yaml.round_trip_dump(config, f)

def export_config(args, output_file=None):
    if not output_file:
        output_file = sys.stdout
    if os.path.exists(args.config):
        with open(args.config, 'rt') as f:
            config = yaml.round_trip_load(f.read())
        STATE['stages'] = config['stages']
        config['config'] = _decrypt_dict(config['config'])
    else:
        config = {
            'stages': {
                env['name']: {
                    'environment': env['name'],
                    'key': 'enter-key-name-here'
                } for env in STATE['awscreds'].environments
            },
            'config': {}}
    if args.json:
        output_file.write(json.dumps(config, indent=4))
    elif config:
        yaml.round_trip_dump(config, output_file)

def _process_section(self, section_value, callback=None, templar=None):
        if not templar:
            templar = self._templar
        processed = ordereddict()
        for key, value in section_value.items():
            if isinstance(value, string_types):
                # strings can be templated
                processed[key] = templar.template(value)
                if isinstance(processed[key], AnsibleUnsafeText):
                    processed[key] = str(processed[key])
            elif isinstance(value, (list, dict)):
                # if it's a dimensional structure, it's cheaper just to serialize
                # it, treat it like a template, and then deserialize it again
                buffer = BytesIO() # use bytes explicitly, not unicode
                yaml.round_trip_dump(value, buffer)
                processed[key] = yaml.round_trip_load(
                    templar.template(buffer.getvalue())
                )
            else:
                # ints, booleans, etc.
                processed[key] = value
            if callback:
                callback(processed)
        return processed

def load_yaml(cls, filename: str):
        """
        Load a hierarchy of sparse objects from a YAML file.

        :param filename: The filename to open.
        :return: The configuration object hierarchy
        """
        def unpack(mapping, parent=None):
            """
            Recursively create Configuration objects with the parent
            correctly set, returning the top-most parent.
            """
            if mapping is None:
                return None

            children = config = None

            if not isinstance(mapping, cls):
                children = mapping.pop('children', None)
                config = cls(**cls._coerce_types(mapping), parent=parent)

            if children is not None and len(children) > 0:
                for child in children:
                    unpack(child, parent=config)
            return config

        if filename in cls._yaml_cache:
            return cls._yaml_cache[filename]

        data = None
        with open(filename, 'r') as yaml_file:
            data = unpack(yaml.round_trip_load(yaml_file.read()))

        if data is not None:
            cls._yaml_cache[filename] = data

        return data

def get_content_from_role(role_name, relative_path):
    role_path = resolve_role_to_path(role_name)
    metadata_file = os.path.join(role_path, relative_path)
    if os.path.exists(metadata_file):
        with open(metadata_file) as ifs:
            metadata = yaml.round_trip_load(ifs)
        return metadata or yaml.compat.ordereddict()
    return yaml.compat.ordereddict()

def set_env(self, env, config=None):
        """
        Loads config from container.yml,  and stores the resulting dict to self._config.

        :param env: string of either 'dev' or 'prod'. Indicates 'dev_overrides' handling.
        :return: None
        """
        assert env in ['dev', 'prod']

        if not config:
            try:
                config = yaml.round_trip_load(open(self.config_path))
            except IOError:
                raise AnsibleContainerNotInitializedException()
            except yaml.YAMLError as exc:
                raise AnsibleContainerConfigException(u"Parsing container.yml - %s" % text_type(exc))

        self._validate_config(config)

        for service, service_config in iteritems(config.get('services') or {}):
            if not service_config or isinstance(service_config, string_types):
                raise AnsibleContainerConfigException(u"Error: no definition found in container.yml for service %s."
                                                      % service)
            self._update_service_config(env, service_config)

        # Insure settings['pwd'] = base_path. Will be used later by conductor to resolve $PWD in volumes.
        if config.get('settings', None) is None:
            config['settings'] = ordereddict()
        config['settings']['pwd'] = self.base_path

        self._resolve_defaults(config)

        logger.debug(u"Parsed config", config=config)
        self._config = config

def _get_variables_from_file(self, var_file):
        """
        Looks for file relative to base_path. If not found, checks relative to base_path/ansible.
        If file extension is .yml | .yaml, parses as YAML, otherwise parses as JSON.

        :return: ruamel.ordereddict
        """
        abspath = path.abspath(var_file)
        if not path.exists(abspath):
            dirname, filename = path.split(abspath)
            raise AnsibleContainerConfigException(
                u'Variables file "%s" not found. (I looked in "%s" for it.)' % (filename, dirname)
            )
        logger.debug("Use variable file: %s", abspath, file=abspath)

        if path.splitext(abspath)[-1].lower().endswith(('yml', 'yaml')):
            try:
                config = yaml.round_trip_load(open(abspath))
            except yaml.YAMLError as exc:
                raise AnsibleContainerConfigException(u"YAML exception: %s" % text_type(exc))
        else:
            try:
                config = json.load(open(abspath))
            except Exception as exc:
                raise AnsibleContainerConfigException(u"JSON exception: %s" % text_type(exc))
        return iteritems(config)

def set_env(self, env, config=None):
        try:
            config = yaml.round_trip_load(open(self.config_path))
        except IOError:
            raise AnsibleContainerNotInitializedException()
        except yaml.YAMLError as exc:
            raise AnsibleContainerConfigException(u"Parsing container.yml - %s" % unicode(exc))

        new_services = yaml.compat.ordereddict()
        for service_name, service_config in iteritems(config.get('services') or {}):
            if service_config.get('containers'):
                # If containers is defined, convert it to services, and drop any other keys
                for container in service_config['containers']:
                    if not container.get('container_name'):
                        raise AnsibleContainerConfigException(
                            u"Expecting container to have container_name defined. None found."
                        )
                    new_service_name = "{}-{}".format(service_name, container['container_name'])
                    new_services[new_service_name] = copy.deepcopy(container)
            else:
                new_services[service_name] = copy.deepcopy(service_config)

        config['services'] = new_services
        super(AnsibleContainerConfig, self).set_env(env, config=config)

        if self._config.get('volumes'):
            for vol_key in self._config['volumes']:
                if 'docker' in self._config['volumes'][vol_key]:
                    settings = copy.deepcopy(self._config['volumes'][vol_key][self.engine_name])
                    self._config['volumes'][vol_key] = settings
                else:
                    # remove non-docker settings
                    for engine_name in self.remove_engines:
                        if engine_name in self._config['volumes'][vol_key]:
                            del self._config['volumes'][vol_key][engine_name]

def _load_config(filename=DOT_CONFIG_YML, roundtrip=False, fixup=True):
    cfg = {}
    if os.path.isfile(filename):
        try:
            with open(filename, 'r') as f:
                if roundtrip:
                    cfg = yaml.round_trip_load(f.read())
                else:
                    cfg = yaml.safe_load(f.read())
            if fixup:
                cfg = _fixup(cfg)
        except Exception as ex:
            print('ex =', ex)
            raise ConfigLoadError(filename, errors=[ex])
    return AttrDict(cfg)