def read(self, stream):
"""
Read the given stream and returns it as a dict.
:param stream: The stream to read the configuration from.
:return IgnoreCaseDict: The configuration read from the stream.
"""
if stream is None:
raise ValueError('stream cannot be None')
return yaml.load(stream, Loader=self._get_loader())
python类Loader()的实例源码
def loads(content):
return yaml.load(content.decode('utf-8'),
Loader=Loader)
def __init__(self, *args, **kwargs):
yaml.Loader.__init__(self, *args, **kwargs)
self.add_constructor(u'tag:yaml.org,2002:map', type(self).construct_yaml_map)
self.add_constructor(u'tag:yaml.org,2002:omap', type(self).construct_yaml_map)
def read_yaml(file: str):
with open(file, 'rU', encoding="utf-8") as stream:
return yaml.load(stream, Loader=Loader)
def __init__(self, *args, **kwargs):
yaml.Loader.__init__(self, *args, **kwargs)
self.add_constructor(u'tag:yaml.org,2002:map', type(self).construct_yaml_map)
self.add_constructor(u'tag:yaml.org,2002:omap', type(self).construct_yaml_map)
def load_config(path):
# TODO: support old-style setting names? i.e. pass them through ARGS_TO_SETTINGS ?
data = {}
yaml_err = ""
if useYAML:
try:
# this assumes only one document
with open(path, encoding='utf8') as infp:
data = yaml_load(infp, Loader=YAMLLoader)
except FileNotFoundError:
return {}
except YAMLError as e:
yaml_err = 'YAML parsing error in file {}'.format(path)
if hasattr(e, 'problem_mark'):
mark = e.problem_mark
yaml_err + '\nError on Line:{} Column:{}'.format(mark.line + 1, mark.column + 1)
else:
return _convert_old_config(data)
try:
with open(path, encoding='utf8') as infp:
data = json.load(infp)
except FileNotFoundError:
return {}
except json.JSONDecodeError as e:
if useYAML and yaml_err:
print(yaml_err)
else:
print('JSON parsing error in file {}'.format(path),
'Error on Line: {} Column: {}'.format(e.lineno, e.colno), sep='\n')
data = _convert_old_config(data)
# if 'directory' in data:
# for k, v in data['directory'].items():
# data['directory'][k] = os.path.expanduser(v)
return data
def load(f):
return yaml.load(f, Loader=FancyLoader)
def yaml_load_ordered(stream, Loader=yaml.Loader, object_pairs_hook=OrderedDict):
class OrderedLoader(Loader):
pass
def construct_mapping(loader, node):
loader.flatten_mapping(node)
return object_pairs_hook(loader.construct_pairs(node))
OrderedLoader.add_constructor(
yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
construct_mapping)
return yaml.load(stream, OrderedLoader)
def __init__(self, stream):
yaml.Loader.__init__(self, stream)
def get_single_data(self):
data = yaml.Loader.get_single_data(self)
return self.recursive_extend(data)
def __init__(self, stream):
self._root = os.path.split(stream.name)[0]
super(Loader, self).__init__(stream)
def include(self, node):
# include another yaml file by `!include` directive
filename = os.path.join(self._root, self.construct_scalar(node))
with open(filename, 'r') as f:
return yaml.load(f, Loader)
def __init__(self, rule_name, metadata):
"""Constructor of TreeLoader.
Args:
rule_name: rule file name in ./rules, excluding extension .yml
metadata: data of any other information needed evaluating tree
"""
base_dir = os.path.dirname(os.path.abspath(__file__))
# Open yaml rule file
try:
f = open(os.path.join(base_dir, 'rules', rule_name + '.yml'), 'r', encoding='utf-8')
self.tree = yaml.load(f, Loader)
f.close()
except IOError:
logger.error('Error opening file: ' + rule_name)
raise TreeLoaderException()
# Metadata
self.metadata = metadata
self.default_properties = {
# if this value is True, falsy node will be hidden
'hide_false': False,
# [currently acquired credit, required credit for True, sum credit of falsy node or not]
'credit_info': [0, 0, False],
'main_node': False,
}
# base node is GRADUATE
self.base_node = TreeNode(None, self.default_properties, self.metadata, and_func(), '!GRADUATE')
# initiate recursive tree loading procedure
self.load_tree(self.tree, self.base_node)
def __init__(self, *args, **kwargs):
yaml.Loader.__init__(self, *args, **kwargs)
toplevel = True
m = u'tag:yaml.org,2002:map'
self.prev_constructor = self.yaml_constructors[m]
self.add_constructor(m, type(self).construct_yaml_map)
def __init__(self, stream):
"""Initialise Loader."""
try:
self._root = os.path.split(stream.name)[0]
except AttributeError:
self._root = os.path.curdir
super().__init__(stream)
def construct_include(self, node):
"""Include file referenced at node."""
filename = os.path.abspath(os.path.join(
self._root, self.construct_scalar(node)
))
filename = os.path.join(self._root, self.construct_scalar(node))
extension = os.path.splitext(filename)[1].lstrip('.')
with open(filename, 'r') as f:
if extension in ('yaml', 'yml'):
return yaml.load(f, Loader)
else:
return ''.join(f.readlines())
def load(filename):
with open(filename, 'r') as f:
data = yaml.load(f, Loader)
# pp = pprint.PrettyPrinter(indent=4)
# pp.pprint(data)
return data
def load(*args, **kwargs):
"""Delegate to yaml load.
"""
if kwargs is None:
kwargs = {}
kwargs['Loader'] = Loader
return yaml.load(*args, **kwargs)
def load_all(*args, **kwargs):
"""Delegate to yaml loadall.
"""
if kwargs is None:
kwargs = {}
kwargs['Loader'] = Loader
return yaml.load_all(*args, **kwargs)
def load(stream, safe=False, many=False):
Loader = SafeCustomLoader if safe else CustomLoader
if many:
return tuple(yaml.load_all(stream, Loader))
else:
return yaml.load(stream, Loader)
def HandleEvent(self, event, loader=None):
"""Handle individual PyYAML event.
Args:
event: Event to forward to method call in method call.
Raises:
IllegalEvent when receives an unrecognized or unsupported event type.
"""
if event.__class__ not in _EVENT_METHOD_MAP:
raise yaml_errors.IllegalEvent(
"%s is not a valid PyYAML class" % event.__class__.__name__)
if event.__class__ in self._event_method_map:
self._event_method_map[event.__class__](event, loader)
def HandleEvent(self, event, loader=None):
"""Handle individual PyYAML event.
Args:
event: Event to forward to method call in method call.
Raises:
IllegalEvent when receives an unrecognized or unsupported event type.
"""
if event.__class__ not in _EVENT_METHOD_MAP:
raise yaml_errors.IllegalEvent(
"%s is not a valid PyYAML class" % event.__class__.__name__)
if event.__class__ in self._event_method_map:
self._event_method_map[event.__class__](event, loader)
def HandleEvent(self, event, loader=None):
"""Handle individual PyYAML event.
Args:
event: Event to forward to method call in method call.
Raises:
IllegalEvent when receives an unrecognized or unsupported event type.
"""
if event.__class__ not in _EVENT_METHOD_MAP:
raise yaml_errors.IllegalEvent(
"%s is not a valid PyYAML class" % event.__class__.__name__)
if event.__class__ in self._event_method_map:
self._event_method_map[event.__class__](event, loader)
def HandleEvent(self, event, loader=None):
"""Handle individual PyYAML event.
Args:
event: Event to forward to method call in method call.
Raises:
IllegalEvent when receives an unrecognized or unsupported event type.
"""
if event.__class__ not in _EVENT_METHOD_MAP:
raise yaml_errors.IllegalEvent(
"%s is not a valid PyYAML class" % event.__class__.__name__)
if event.__class__ in self._event_method_map:
self._event_method_map[event.__class__](event, loader)
def HandleEvent(self, event, loader=None):
"""Handle individual PyYAML event.
Args:
event: Event to forward to method call in method call.
Raises:
IllegalEvent when receives an unrecognized or unsupported event type.
"""
if event.__class__ not in _EVENT_METHOD_MAP:
raise yaml_errors.IllegalEvent(
"%s is not a valid PyYAML class" % event.__class__.__name__)
if event.__class__ in self._event_method_map:
self._event_method_map[event.__class__](event, loader)
def StreamStart(self, event, loader):
"""Handle start of stream event"""
def StreamEnd(self, event, loader):
"""Handle end of stream event"""
def DocumentStart(self, event, loader):
"""Handle start of document event"""
def DocumentEnd(self, event, loader):
"""Handle end of document event"""
def Alias(self, event, loader):
"""Handle alias event"""
