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

def generate_cwl_documentation(_):
    cur_dir = os.path.abspath(os.path.dirname(__file__))

    # find all cwl files
    with WorkflowGenerator() as wf:
        cwl_files = [step.run for step in wf.steps_library.steps.values()]
    # sort alphabetically
    cwl_files.sort()

    tools_file = os.path.join(cur_dir, 'tools.rst')
    tool_template = '\n{}\n{}\n\n{}\n'
    with codecs.open(tools_file, 'wb', encoding='utf-8') as f:
        f.write('Tools\n=====\n')
        f.write('\n``nlppln`` contains the following tools:\n')
        for cwl in cwl_files:
            tool_name = os.path.basename(cwl)
            plusses = '+'*len(tool_name)
            with codecs.open(cwl) as c:
                try:
                    cwl_yaml = yaml.load(c, Loader=yaml.RoundTripLoader)
                    doc = cwl_yaml.get('doc', 'No documentation')
                    f.write(tool_template.format(tool_name, plusses, doc))
                except yaml.YAMLError:
                    pass

def save(self, path: str):
        cfg = self._dict_to_orderdict(self.cfg_dict)
        with open(path[:-3] + 'yaml', 'w') as cfg_file:
            try:
                yaml3ed.dump(cfg, cfg_file, explicit_start=True, explicit_end=True,
                             default_flow_style=False, allow_unicode=True, version=(1, 2),
                             indent=2)
            except yaml3ed.YAMLError as exc:
                print(exc)

def load(self, path_cfg: str):
        with open(path_cfg, 'r') as stream:
            try:
                self.cfg_dict = yaml3ed.safe_load(stream)
            except yaml3ed.YAMLError as exc:
                print(exc)
        self.check()
        self.zbx = ZabbixAgent(self.cfg_dict['zabbix']['url'], self.cfg_dict['zabbix']['login'],
                               self.cfg_dict['zabbix']['password'])
        log.debug('Config loaded')

def save(self, path: str):
        cfg = self._dict_to_orderdict(self.map_config)
        with open(path + '/' + self.map_data['name'] + '.yaml', 'w') as cfg_file:
            try:
                yaml3ed.dump(cfg, cfg_file, explicit_start=True, explicit_end=True,
                             default_flow_style=False, allow_unicode=True, version=(1, 2))
            except yaml3ed.YAMLError as exc:
                print(exc)

def load_yaml(self, template_file):
        with open(template_file) as f:
            try:
                return yaml.safe_load(f)
            except yaml.YAMLError as e:
                print(e)
                return []

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(yaml_string, schema=None, label=u"<unicode string>"):
    """
    Parse the first YAML document in a string
    and produce corresponding YAML object.
    """
    if str(type(yaml_string)) not in ("<type 'unicode'>", "<type 'str'>", "<class 'str'>"):
        raise TypeError("StrictYAML can only read a string of valid YAML.")

    # We manufacture a class that has the label we want
    DynamicStrictYAMLLoader = type('DynamicStrictYAMLLoader', (StrictYAMLLoader,), {"label": label})

    try:
        document = ruamelyaml.load(yaml_string, Loader=DynamicStrictYAMLLoader)
    except ruamelyaml.YAMLError as parse_error:
        if parse_error.context_mark is not None:
            parse_error.context_mark.name = label
        if parse_error.problem_mark is not None:
            parse_error.problem_mark.name = label

        raise parse_error

    # Document is just a (string, int, etc.)
    if type(document) not in (CommentedMap, CommentedSeq):
        document = yaml_string

    if schema is None:
        schema = Any()

    return schema(YAMLChunk(document, label=label))

def parse(filename):
        """
        Parse a YAML file
        """
        try:
            with open(filename) as f:
                try:
                    return yaml.load(f)
                except yaml.YAMLError as e:
                    log.critical("Problem parsing {} as YAML: {}".format(
                        filename, e))
                    return None
        except FileNotFoundError:
            log.critical("Problem opening {}: File was not found".format(filename))
            return None

def __init__(self, *args, **kwargs):
        if type(self) is BaseProvider:
            return
        if self.debug:
            print(f'{self._typ.upper()} Provider .ctor')

        if not self._base_instance:
            self._base_instance = self.__class__.__bases__[0]()
        base_attributes = inspect.getmembers(self._base_instance, lambda a:not(inspect.isroutine(a)))
        base_keys = [a[0] for a in base_attributes if not(a[0].startswith('_'))]
        # base_keys = ['debug', 'default_mc_version']

        for attribute_key in base_keys:
            if attribute_key in kwargs:
                value = kwargs.get(attribute_key)
                setattr(self, attribute_key, kwargs[attribute_key])

        provider_settings = kwargs.get('provider_settings', {})
        provider_settings = provider_settings.get(self._typ, {})

        if self.debug:
            print(f'{self._typ} settings: {provider_settings}')

        attributes = inspect.getmembers(self, lambda a:not(inspect.isroutine(a)))

        attribute_keys = [a[0] for a in attributes if not(a[0].startswith('_'))]
        attribute_keys = list(set(attribute_keys) - set(base_keys))

        path = Path(kwargs['data_path'], 'defaults.yaml')
        if path.is_dir():
            path.rmdir()
        global_defaults = {}
        if path.exists():
            with open(path, 'r') as stream:
                try:
                    global_defaults = yaml.safe_load(stream)
                except yaml.YAMLError as exc:
                    print(exc)
                    global_defaults = {}
        # get all default values
        global_defaults[self._typ] = {k: getattr(self, k) for k in attribute_keys}
        with open(path, 'w') as outfile:
            yaml.dump(global_defaults, outfile, default_flow_style=False)

        # write provider settings overrides to self
        for attribute_key in attribute_keys:
            if attribute_key in provider_settings:
                value = provider_settings.get(attribute_key)
                if self.debug:
                    print(f'setting {attribute_key}, value={value}')
                setattr(self, attribute_key, provider_settings[attribute_key])