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
python类MarkedYAMLError()的实例源码
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
