我们从Python开源项目中，提取了以下26个代码示例，用于说明如何使用yaml.MappingNode()。
def read_configuration(cls, config_file): """read YAML configuration file""" # load YAML events/measurements definition f = open(config_file, 'r') doc_yaml = yaml.compose(f) f.close() # split events & measurements definitions measurements, events = list(), list() for key, value in doc_yaml.value: if value.tag == Measurements.yaml_tag: measurements.append((key, value)) if value.tag == Events.yaml_tag: events.append((key, value)) measurements_yaml = yaml.MappingNode(u'tag:yaml.org,2002:map', measurements) measurements_stream = yaml.serialize(measurements_yaml) events_yaml = yaml.MappingNode(u'tag:yaml.org,2002:map', events) events_stream = yaml.serialize(events_yaml) # return event & measurements definition return events_stream, measurements_stream
def construct_mapping(self, node, deep=False): if isinstance(node, yaml.MappingNode): self.flatten_mapping(node) else: raise yaml.constructor.ConstructorError(None, None, 'expected a mapping node, but found %s' % node.id, node.start_mark) mapping = OrderedDict() for key_node, value_node in node.value: key = self.construct_object(key_node, deep=deep) try: hash(key) except TypeError, exc: raise yaml.constructor.ConstructorError('while constructing a mapping', node.start_mark, 'found unacceptable key (%s)' % exc, key_node.start_mark) value = self.construct_object(value_node, deep=deep) mapping[key] = value return mapping
def represent_odict(dump, tag, mapping, flow_style=None): """ Like BaseRepresenter.represent_mapping, but does not issue the sort(). """ value = [] node = yaml.MappingNode(tag, value, flow_style=flow_style) if dump.alias_key is not None: dump.represented_objects[dump.alias_key] = node best_style = True if hasattr(mapping, 'items'): mapping = mapping.items() for item_key, item_value in mapping: node_key = dump.represent_data(item_key) node_value = dump.represent_data(item_value) if not (isinstance(node_key, yaml.ScalarNode) and not node_key.style): best_style = False if not (isinstance(node_value, yaml.ScalarNode) and not node_value.style): best_style = False value.append((node_key, node_value)) if flow_style is None: if dump.default_flow_style is not None: node.flow_style = dump.default_flow_style else: node.flow_style = best_style return node
def represent_mapping(self, tag, mapping, flow_style=None): value = [] node = yaml.MappingNode(tag, value, flow_style=flow_style) if self.alias_key is not None: self.represented_objects[self.alias_key] = node best_style = False if hasattr(mapping, 'items'): mapping = list(mapping.items()) for item_key, item_value in mapping: node_key = self.represent_data(item_key) node_value = self.represent_data(item_value) if not (isinstance(node_key, yaml.ScalarNode) and not node_key.style): best_style = False if not (isinstance(node_value, yaml.ScalarNode) and not node_value.style): best_style = False value.append((node_key, node_value)) if flow_style is None: if self.default_flow_style is not None: node.flow_style = self.default_flow_style else: node.flow_style = best_style return node
def construct_mapping(self, node, deep=False): if isinstance(node, yaml.MappingNode): self.flatten_mapping(node) else: raise yaml.constructor.ConstructorError(None, None, 'expected a mapping node, but found %s' % node.id, node.start_mark) mapping = OrderedDict() for key_node, value_node in node.value: key = self.construct_object(key_node, deep=deep) try: hash(key) except TypeError as exc: raise yaml.constructor.ConstructorError('while constructing a mapping', node.start_mark, 'found unacceptable key (%s)' % exc, key_node.start_mark) value = self.construct_object(value_node, deep=deep) mapping[key] = value return mapping
def construct_mapping(self, node, deep=False): if isinstance(node, yaml.MappingNode): self.flatten_mapping(node) else: raise yaml.constructor.ConstructorError(None, None, 'expected a mapping node, but found %s' % node.id, node.start_mark) mapping = OrderedDict() for key_node, value_node in node.value: key = self.construct_object(key_node, deep=deep) try: hash(key) except TypeError as exc: raise yaml.constructor.ConstructorError('while constructing a mapping', node.start_mark, 'found unacceptable key (%s)' % exc, key_node.start_mark) value = self.construct_object(value_node, deep=deep) if key in mapping: raise KeyError('There are two same keys "'+node+'" in the yaml file.') mapping[key] = value return mapping
def multi_constructor(loader, tag_suffix, node): """ Deal with !Ref style function format """ if tag_suffix not in UNCONVERTED_SUFFIXES: tag_suffix = "{}{}".format(FN_PREFIX, tag_suffix) constructor = None if tag_suffix == "Fn::GetAtt": constructor = construct_getatt elif isinstance(node, yaml.ScalarNode): constructor = loader.construct_scalar elif isinstance(node, yaml.SequenceNode): constructor = loader.construct_sequence elif isinstance(node, yaml.MappingNode): constructor = loader.construct_mapping else: raise "Bad tag: !{}".format(tag_suffix) return ODict(( (tag_suffix, constructor(node)), ))
def represent_odict(dump, tag, mapping, flow_style=None): """Like BaseRepresenter.represent_mapping, but does not issue the sort(). """ value = [] node = yaml.MappingNode(tag, value, flow_style=flow_style) if dump.alias_key is not None: dump.represented_objects[dump.alias_key] = node best_style = True if hasattr(mapping, 'items'): mapping = mapping.items() for item_key, item_value in mapping: node_key = dump.represent_data(item_key) node_value = dump.represent_data(item_value) if not (isinstance(node_key, yaml.ScalarNode) and not node_key.style): best_style = False if not (isinstance(node_value, yaml.ScalarNode) and not node_value.style): best_style = False value.append((node_key, node_value)) if flow_style is None: if dump.default_flow_style is not None: node.flow_style = dump.default_flow_style else: node.flow_style = best_style return node
def represent_mapping(self, tag, mapping, flow_style=None): value = [] node = MappingNode(tag, value, flow_style=flow_style) if self.alias_key is not None: self.represented_objects[self.alias_key] = node best_style = True if hasattr(mapping, 'items'): mapping = list(mapping.items()) if not isinstance(mapping, OrderedDict): mapping.sort() for item_key, item_value in mapping: node_key = self.represent_data(item_key) node_value = self.represent_data(item_value) if not (isinstance(node_key, ScalarNode) and not node_key.style): best_style = False if not (isinstance(node_value, ScalarNode) and not node_value.style): best_style = False value.append((node_key, node_value)) if flow_style is None: if self.default_flow_style is not None: node.flow_style = self.default_flow_style else: node.flow_style = best_style return node
def construct_mapping(cls, loader, node, deep=False): """Based on yaml.loader.BaseConstructor.construct_mapping.""" if not isinstance(node, yaml.MappingNode): raise yaml.loader.ConstructorError( None, None, "expected a mapping node, but found %s" % node.id, node.start_mark) mapping = OrderedYamlDict() for key_node, value_node in node.value: key = loader.construct_object(key_node, deep=deep) try: hash(key) except TypeError, exc: raise yaml.loader.ConstructorError( "while constructing a mapping", node.start_mark, "found unacceptable key (%s)" % exc, key_node.start_mark) value = loader.construct_object(value_node, deep=deep) mapping[key] = value return mapping
def construct_mapping(cls, loader, node, deep=False): """Based on yaml.loader.BaseConstructor.construct_mapping.""" if not isinstance(node, yaml.MappingNode): raise yaml.loader.ConstructorError( None, None, "expected a mapping node, but found %s" % node.id, node.start_mark) mapping = OrderedYamlDict() for key_node, value_node in node.value: key = loader.construct_object(key_node, deep=deep) try: hash(key) except TypeError as exc: raise yaml.loader.ConstructorError( "while constructing a mapping", node.start_mark, "found unacceptable key (%s)" % exc, key_node.start_mark) value = loader.construct_object(value_node, deep=deep) mapping[key] = value return mapping
def construct_mapping(self, node, deep=None): if isinstance(node, yaml.MappingNode): self.flatten_mapping(node) else: if hasattr(node, 'id') and hasattr(node, 'start_mark'): error = 'expected a mapping node, but found {}'.format(node.id) raise yaml.constructorConstructorError(None, None, error, node.start_mark) else: error = 'Invalid type for node variable: {} - {}'.format(type(node), node) raise TypeError(error) mapping = self.container() for key_node, value_node in node.value: key = self.construct_object(key_node, deep=deep) try: hash(key) except TypeError as e: raise yaml.constructorConstructorError( 'while constructing a mapping', node.start_mark, 'found unacceptable key ({})'.format(e), key_node.start_mark ) value = self.construct_object(value_node, deep=deep) mapping[key] = value return mapping
def format_node(cls, mapping, metric): if mapping.tag in [ 'tag:yaml.org,2002:str', Bytes2Kibibytes.yaml_tag, Number.yaml_tag, StripExtraDash.yaml_tag]: return yaml.ScalarNode(mapping.tag, mapping.value.format(**metric)) elif mapping.tag == 'tag:yaml.org,2002:map': values = [] for key, value in mapping.value: values.append((yaml.ScalarNode(key.tag, key.value), cls.format_node(value, metric))) return yaml.MappingNode(mapping.tag, values) elif mapping.tag in [ArrayItem.yaml_tag, ValueItem.yaml_tag]: values = [] for seq in mapping.value: map_values = list() for key, value in seq.value: if key.value == 'SELECT': map_values.append((yaml.ScalarNode(key.tag, key.value), cls.format_node(value, metric))) else: map_values.append((yaml.ScalarNode(key.tag, key.value), value)) values.append(yaml.MappingNode(seq.tag, map_values)) return yaml.SequenceNode(mapping.tag, values) elif mapping.tag in [MapValue.yaml_tag]: values = [] for key, value in mapping.value: if key.value == 'VALUE': values.append((yaml.ScalarNode(key.tag, key.value), cls.format_node(value, metric))) else: values.append((yaml.ScalarNode(key.tag, key.value), value)) return yaml.MappingNode(mapping.tag, values) return mapping
def construct_mapping(self, node, deep=False): if isinstance(node, yaml.MappingNode): self.flatten_mapping(node) else: raise yaml.constructor.ConstructorError( None, None, u'expected a mapping node, but found %s' % node.id, node.start_mark ) mapping = OrderedDict() for key_node, value_node in node.value: key = self.construct_object(key_node, deep=deep) try: hash(key) except TypeError as exc: raise yaml.constructor.ConstructorError( u'while constructing a mapping', node.start_mark, 'found unacceptable key (%s)' % exc, key_node.start_mark ) value = self.construct_object(value_node, deep=deep) mapping[key] = value return mapping # Allow bare strings to begin with %. Directives are still detected.
def represent_ordereddict(dumper, data): value = [] for item_key, item_value in data.items(): node_key = dumper.represent_data(item_key) node_value = dumper.represent_data(item_value) value.append((node_key, node_value)) return yaml.nodes.MappingNode(u'tag:yaml.org,2002:map', value) # make ordered dicts dump as normal dicts
def construct_mapping(self, node, deep=False): if isinstance(node, yaml.MappingNode): self.flatten_mapping(node) else: raise yaml.constructor.ConstructorError( None , None , 'expected a mapping node, but found {}'.format(node.id) , node.start_mark ) mapping = collections.OrderedDict() for key_node, value_node in node.value: key = self.construct_object(key_node, deep=deep) try: hash(key) except TypeError as ex: raise yaml.constructor.ConstructorError( 'while constructing a mapping' , node.start_mark , 'found unacceptable key `{}`'.format(ex) , key_node.start_mark ) value = self.construct_object(value_node, deep=deep) mapping[key] = value return mapping
def generic_object(loader, suffix, node): if isinstance(node, yaml.ScalarNode): constructor = loader.__class__.construct_scalar elif isinstance(node, yaml.SequenceNode): constructor = loader.__class__.construct_sequence elif isinstance(node, yaml.MappingNode): constructor = loader.__class__.construct_mapping else: raise ValueError(node) # TODO(tailhook) wrap into some object? return constructor(loader, node)
def from_yaml(cls, loader, node): key = node.tag[1:] if node.tag not in ('!Ref', '!Condition'): key = 'Fn::' + key if isinstance(node, ScalarNode): val = loader.construct_scalar(node) elif isinstance(node, SequenceNode): val = loader.construct_sequence(node) elif isinstance(node, MappingNode): val = loader.construct_mapping(node) else: raise Exception("Unable to handle node: %r"%node) return {str(key): str(val)}
def construct_odict(self, node): omap = OrderedDict() yield omap if not isinstance(node, yaml.SequenceNode): raise yaml.constructor.ConstructorError( "while constructing an ordered map", node.start_mark, "expected a sequence, but found %s" % node.id, node.start_mark ) for subnode in node.value: if not isinstance(subnode, yaml.MappingNode): raise yaml.constructor.ConstructorError( "while constructing an ordered map", node.start_mark, "expected a mapping of length 1, but found %s" % subnode.id, subnode.start_mark ) if len(subnode.value) != 1: raise yaml.constructor.ConstructorError( "while constructing an ordered map", node.start_mark, "expected a single mapping item, but found %d items" % len(subnode.value), subnode.start_mark ) key_node, value_node = subnode.value[0] key = self.construct_object(key_node) value = self.construct_object(value_node) omap[key] = value
def to_yaml(cls, dumper, data): value = [] node = yaml.nodes.MappingNode(cls.yaml_tag, value) for key, item in data.iteritems(): node_key = dumper.represent_data(key) node_value = dumper.represent_data(item) value.append((node_key, node_value)) return node
def represent_orderedyamldict(dumper, data): value = [] for item_key, item_value in data.items(): node_key = dumper.represent_data(item_key) if type(item_value) not in [ str, unicode, list, dict, OrderedYamlDict, bool, long, int]: import pdb; pdb.set_trace() node_value = dumper.represent_data(item_value) value.append((node_key, node_value)) return yaml.nodes.MappingNode(u'tag:yaml.org,2002:map', value)
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 from_yaml(cls, loader, node): logging.debug('{}:process(loader={}, node={})'.format(cls.__name__, loader, node)) # e.g.: # SequenceNode(tag=u'!ArrayItem', value=[ # MappingNode(tag=u'tag:yaml.org,2002:map', value=[ # (ScalarNode(tag=u'tag:yaml.org,2002:str', value=u'SELECT'), # MappingNode(tag=u'tag:yaml.org,2002:map', value=[ # (ScalarNode(tag=u'tag:yaml.org,2002:str', value=u'plugin'), # , ...) # ]), ... # ), (key, value), ... ]) # , ... ]) assert isinstance(node, yaml.SequenceNode), \ "{} tag isn't YAML array".format(cls.__name__) select, index_keys, items, item_desc = list(), list(), list(), None for elem in node.value: for key, value in elem.value: if key.value == 'ITEM-DESC': assert item_desc is None, "ITEM-DESC key already set" item_desc = value if key.value == 'INDEX-KEY': assert len(index_keys) == 0, "INDEX-KEY key already set" index_keys = loader.construct_sequence(value) if key.value == 'SELECT': select.append(loader.construct_mapping(value)) # validate item description assert item_desc is not None, "Mandatory ITEM-DESC key isn't set" assert len(select) > 0 or len(index_keys) > 0, \ "Mandatory key (INDEX-KEY or SELECT) isn't set" metrics = loader.collector.items(select) # select metrics based on INDEX-KEY provided if len(index_keys) > 0: metric_set = set() for metric in metrics: value = CollectdValue() for key in index_keys: setattr(value, key, getattr(metric, key)) metric_set.add(value) metrics = list(metric_set) # build items based on SELECT and/or INDEX-KEY criteria for metric in metrics: item = cls.format_node(item_desc, {'vl': metric, 'system': loader.system, 'config': loader.config}) items.append(loader.construct_mapping(item)) return items
def _get_loader(self): """ Get a loader that uses an IgnoreCaseDict for complex objects. :return yaml.Loader: The loader object. """ # this class was copied from # https://github.com/fmenabe/python-yamlordereddictloader/blob/master/yamlordereddictloader.py # and adapted to use IgnoreCaseDict class Loader(yaml.Loader): def __init__(self, *args, **kwargs): yaml.Loader.__init__(self, *args, **kwargs) self.add_constructor( 'tag:yaml.org,2002:map', type(self).construct_yaml_map) self.add_constructor( 'tag:yaml.org,2002:omap', type(self).construct_yaml_map) def construct_yaml_map(self, node): data = IgnoreCaseDict() yield data value = self.construct_mapping(node) data.update(value) def construct_mapping(self, node, deep=False): if isinstance(node, yaml.MappingNode): self.flatten_mapping(node) else: raise yaml.constructor.ConstructorError( None, None, 'expected a mapping node, but found %s' % node.id, node.start_mark) mapping = IgnoreCaseDict() for key_node, value_node in node.value: key = self.construct_object(key_node, deep=deep) try: hash(key) except TypeError as err: raise yaml.constructor.ConstructorError( 'while constructing a mapping', node.start_mark, 'found unacceptable key (%s)' % err, key_node.start_mark) value = self.construct_object(value_node, deep=deep) mapping[key] = value return mapping return Loader
