Python yaml 模块，MarkedYAMLError() 实例源码

def load(self):
        # Prepare + load directory.
        super().load()

        # Load the files and parse Yaml.
        parsed_settings = dict()

        try:
            for file_name in self.files:
                file_path = os.path.join(self.directory, file_name)
                with open(file_path, 'r') as file_handle:
                    parsed_settings.update(yaml.load(file_handle))
        except (yaml.YAMLError, yaml.MarkedYAMLError) as e:
            raise ImproperlyConfigured(
                'Your settings file(s) contain invalid YAML syntax! Please fix and restart!, {}'.format(str(e))
            )

        # Loop and set in local settings (+ uppercase keys).
        for key, value in parsed_settings.items():
            self.settings[key.upper()] = value

def _ordered_dict(loader: SafeLineLoader,
                  node: yaml.nodes.MappingNode) -> OrderedDict:
    """Load YAML mappings into an ordered dictionary to preserve key order."""
    loader.flatten_mapping(node)
    nodes = loader.construct_pairs(node)

    seen = {}  # type: Dict
    for (key, _), (child_node, _) in zip(nodes, node.value):
        line = child_node.start_mark.line

        try:
            hash(key)
        except TypeError:
            fname = getattr(loader.stream, 'name', '')
            raise yaml.MarkedYAMLError(
                context="invalid key: \"{}\"".format(key),
                context_mark=yaml.Mark(fname, 0, line, -1, None, None)
            )

        if key in seen:
            fname = getattr(loader.stream, 'name', '')
            first_mark = yaml.Mark(fname, 0, seen[key], -1, None, None)
            second_mark = yaml.Mark(fname, 0, line, -1, None, None)
            raise yaml.MarkedYAMLError(
                context="duplicate key: \"{}\"".format(key),
                context_mark=first_mark, problem_mark=second_mark,
            )
        seen[key] = line

    processed = OrderedDict(nodes)
    setattr(processed, '__config_file__', loader.name)
    setattr(processed, '__line__', node.start_mark.line)
    return processed

def _read_config(paths, testing=False):
    """
    Read the config files listed in path and merge them into a dictionary.

    :arg paths: filenames to read and merge configuration data from.  The last
        files should override the first ones.
    :arg testing: If set to True, then use paths to data files that are
        appropriate for testing a source checkout.
    :rtype: ConfigObj, a dict-like object with helper methods for use as a config store
    :returns: Return the configuration dict
    """
    cfg = yaml.safe_load(DEFAULT_CONFIG)
    CONFIG_SCHEMA(cfg)

    for cfg_file in paths:
        try:
            with open(cfg_file, 'rb') as f:
                new_cfg = yaml.safe_load(f)
        except yaml.MarkedYAMLError as e:
            raise MagnateConfigError('Config error parsing {}:\n{}'.format(cfg_file, e))

        _merge_mapping(cfg, new_cfg, inplace=True)

        try:
            CONFIG_SCHEMA(cfg)
        except MultipleInvalid as e:
            raise MagnateConfigError('Config error in {}:\n{}'.format(cfg_file, e))

    if testing:
        testing_cfg = yaml.safe_load(TESTING_CONFIG)
        _merge_mapping(cfg, testing_cfg, inplace=True)
        CONFIG_SCHEMA(cfg)

    return cfg

def yaml_error(loader, e, exc_info=None):
        # Given a MarkedYAMLError exception, either log or raise
        # the error, depending on the 'fatal' argument.
        fatal = not getattr(loader, 'warnings', False)
        setattr(loader, 'yaml_errored', True)

        if exc_info:
            # When we're given the original exception (which was wrapped in
            # a MarkedYAMLError), we can provide more context for debugging.

            from traceback import format_exc
            e.note = "\nOriginal exception:\n" + format_exc(exc_info)

        if fatal:
            raise e

        message = []

        mark = e.context_mark or e.problem_mark
        if mark:
            position = "%s:%s:%s" % (mark.name, mark.line+1, mark.column+1)
        else:
            position = "[unknown]"
        if e.context is not None:
            message.append(e.context)

        if e.problem is not None:
            message.append(e.problem)

        if e.note is not None:
            message.append("(note: " + e.note + ")")

        print "%s: %s" % (position, ",".join(message))

def construct_specsparameters(loader, node, spectype):
        spec_class = {x.__name__: x for x in Spec.__subclasses__()}.get(spectype, None)

        if not spec_class:
            yaml_error(loader, yaml.constructor.ConstructorError(
                None, None,
                "Unrecognized Spec class %s" % spectype,
                node.start_mark))
            return

        if not isinstance(node, yaml.MappingNode):
            yaml_error(loader, yaml.constructor.ConstructorError(
                None, None,
                "expected a mapping node, but found %s" % node.id,
                node.start_mark))
            return

        specs = OrderedDict()
        for spec_key_node, spec_value_node in node.value:
            try:
                spec_key = str(loader.construct_scalar(spec_key_node))
            except yaml.MarkedYAMLError, e:
                yaml_error(loader, e)

            specs[spec_key] = construct_spec(spec_class, loader, spec_value_node)

        return specs

def _ordered_dict(loader: SafeLineLoader,
                  node: yaml.nodes.MappingNode) -> OrderedDict:
    """Load YAML mappings into an ordered dictionary to preserve key order."""
    loader.flatten_mapping(node)
    nodes = loader.construct_pairs(node)

    seen = {}  # type: Dict
    for (key, _), (child_node, _) in zip(nodes, node.value):
        line = child_node.start_mark.line

        try:
            hash(key)
        except TypeError:
            fname = getattr(loader.stream, 'name', '')
            raise yaml.MarkedYAMLError(
                context="invalid key: \"{}\"".format(key),
                context_mark=yaml.Mark(fname, 0, line, -1, None, None)
            )

        if key in seen:
            fname = getattr(loader.stream, 'name', '')
            first_mark = yaml.Mark(fname, 0, seen[key], -1, None, None)
            second_mark = yaml.Mark(fname, 0, line, -1, None, None)
            raise yaml.MarkedYAMLError(
                context="duplicate key: \"{}\"".format(key),
                context_mark=first_mark, problem_mark=second_mark,
            )
        seen[key] = line

    processed = OrderedDict(nodes)
    setattr(processed, '__config_file__', loader.name)
    setattr(processed, '__line__', node.start_mark.line)
    return processed