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

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 _dump_yaml(applied):
        """
        Dumps dict correctly processing multiline pre & postDescription string
        :param applied: dict -> filled config ready to be converted to yaml
        :return: str -> yaml config
        """

        def str_presenter(dumper, data):
            # check for multiline strings
            if len(data.splitlines()) == 1 and data[-1] == '\n':
                return dumper.represent_scalar(
                    'tag:yaml.org,2002:str', data, style='>')
            if len(data.splitlines()) > 1:
                return dumper.represent_scalar(
                    'tag:yaml.org,2002:str', data, style='|')
            return dumper.represent_scalar(
                'tag:yaml.org,2002:str', data.strip())

        yaml.add_representer(unicode, str_presenter)
        yaml.add_representer(str, str_presenter)
        return yaml.dump(applied, default_flow_style=False, width=1000)

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)
        self.add_implicit_resolver(
            u'tag:yaml.org,2002:float',
            re.compile(u'''^(?:
             [-+]?(?:[0-9][0-9_]*)\\.[0-9_]*(?:[eE][-+]?[0-9]+)?
            |[-+]?(?:[0-9][0-9_]*)(?:[eE][-+]?[0-9]+)
            |\\.[0-9_]+(?:[eE][-+][0-9]+)?
            |[-+]?[0-9][0-9_]*(?::[0-5]?[0-9])+\\.[0-9_]*
            |[-+]?\\.(?:inf|Inf|INF)
            |\\.(?:nan|NaN|NAN))$''', re.X),
            list(u'-+0123456789.'))
        self.filenames = [os.path.abspath(stream.name)]

def ordered_load(
        stream, Loader=yaml.SafeLoader, object_pairs_hook=OrderedDict):
    """Load yaml as an ordered dict

    This allows us to inspect the order of the file on disk to make
    sure it was correct by our rules.
    """
    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)
    # for parameters.yaml we treat numbers (especially version
    # numbers) as strings. So that microversion specification of 2.20
    # and 2.2 don't get confused.
    OrderedLoader.add_constructor(
        u'tag:yaml.org,2002:float',
        yaml.constructor.SafeConstructor.construct_yaml_str)

    return yaml.load(stream, OrderedLoader)

def represent_stringish(dumper, data):
        # Will crash on bytestrings with weird chars in them,
        #  because we can't tell if it's supposed to be e.g. utf-8 readable string
        #  or an arbitrary binary buffer, and former one *must* be pretty-printed
        # PyYAML's Representer.represent_str does the guesswork and !!binary or !!python/str
        # Explicit crash on any bytes object might be more sane, but also annoying
        # Use something like base64 to encode such buffer values instead
        # Having such binary stuff pretty much everywhere on unix (e.g. paths) kinda sucks
        data = unicode(data) # read the comment above

        # Try to use '|' style for multiline data,
        #  quoting it with 'literal' if lines are too long anyway,
        #  not sure if Emitter.analyze_scalar can also provide useful info here
        style = dumper.pyaml_string_val_style
        if not style:
            style = 'plain'
            if '\n' in data or not data or data == '-' or data[0] in '!&*[':
                style = 'literal'
                if '\n' in data[:-1]:
                    for line in data.splitlines():
                        if len(line) > dumper.best_width: break
                    else: style = '|'

        return yaml.representer.ScalarNode('tag:yaml.org,2002:str', data, style=style)

def unicode_representer(_, data):
    has_wide_lines = False
    for line in data.splitlines():
        if len(line) > 80:
            has_wide_lines = True
            break

    if has_wide_lines:
        return yaml.ScalarNode(
            u'tag:yaml.org,2002:str', data, style='>')

    if "\n" in data:
        return yaml.ScalarNode(
            u'tag:yaml.org,2002:str', data, style='|')

    return yaml.ScalarNode(
        u'tag:yaml.org,2002:str', data, style='')

def configure_yml(self):

        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)

        self.collect_answers()
        cfg = self.process_answers()

        yaml.add_representer(OrderedDict, represent_ordereddict)
        return yaml.dump(cfg, default_flow_style=False)

def __init__(self, *args, **kwds):
        yaml.Loader.__init__(self, *args, **kwds)
        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 string_presenter(self, dumper, data):
    """Presenter to force yaml.dump to use multi-line string style."""
    if '\n' in data:
      return dumper.represent_scalar('tag:yaml.org,2002:str', data, style='|')
    else:
      return dumper.represent_scalar('tag:yaml.org,2002:str', data)

def represent_decimal(self, data):
        return self.represent_scalar('tag:yaml.org,2002:str', str(data))

def represent_ordered_dict(self, data):
        return self.represent_mapping('tag:yaml.org,2002:map', data.items())

def represent_decimal(self, data):
        return self.represent_scalar('tag:yaml.org,2002:str', str(data))

def represent_ordered_dict(self, data):
        return self.represent_mapping('tag:yaml.org,2002:map', data.items())

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 represent_decimal(self, data):
        return self.represent_scalar('tag:yaml.org,2002:str', str(data))

def represent_ordered_dict(self, data):
        return self.represent_mapping('tag:yaml.org,2002:map', data.items())

def yaml_load(stream):
    """Load YAML document, but load all strings as unicode on py2."""
    import yaml

    class UnicodeLoader(yaml.SafeLoader):
        """Yaml SafeLoader Class, default encoding is UTF-8."""
        pass
    # NOTE:
    # In [2]: yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG
    # Out[2]: 'tag:yaml.org,2002:map'
    UnicodeLoader.add_constructor(
        yaml.resolver.BaseResolver.DEFAULT_SCALAR_TAG,
        UnicodeLoader.construct_scalar)
    return yaml.load(stream, UnicodeLoader)

def pocketsphinx(self):
        # NOTE: copy()
        # Return a shallow copy of x.
        # source: https://docs.python.org/3/library/copy.html
        """Get pocketsphinx speech to text settings."""
        return self._data.get('pocketsphinx', {}).copy()

def features_enabled(self):
        """Get features settings."""
        return self._data.get('features', [])

# In [2]: yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG
# Out[2]: 'tag:yaml.org,2002:map'

# In [3]:

def _read_yaml(filename):
    yaml.Loader.add_constructor(u'tag:yaml.org,2002:str', construct_yaml_str)
    yaml.SafeLoader.add_constructor(u'tag:yaml.org,2002:str', construct_yaml_str)
    with open(filename, 'r') as f:
        return yaml.safe_load(f.read())

def setup_yaml():
    """ http://stackoverflow.com/a/8661021 """
    represent_dict_order = lambda self, data: self.represent_mapping(
        'tag:yaml.org,2002:map', data.items())
    yaml.add_representer(collections.OrderedDict, represent_dict_order)

def unicode_representer(dumper, uni):
    node = yaml.ScalarNode(tag=u'tag:yaml.org,2002:str', value=str(uni))
    return node

def represent_decimal(self, data):
        return self.represent_scalar('tag:yaml.org,2002:str', str(data))

def represent_ordered_dict(self, data):
        return self.represent_mapping('tag:yaml.org,2002:map', data.items())

def literal_unicode_representer(dumper, data):
    """
    Use |- literal syntax for long strings
    """
    if '\n' in data:
        return dumper.represent_scalar(u'tag:yaml.org,2002:str', data, style='|')
    else:
        return dumper.represent_scalar(u'tag:yaml.org,2002:str', data)

def write_content(self, content=None):
        if content:
            self.content = content
        self.__documents[self.__document_id] = self.content

        def representer(dumper, data):
            """Represents a dict key started with '!' as a YAML tag

            Assumes that there is only one !tag in the dict at the
            current indent.

            Python object:
            {"!unknown_tag": ["some content", ]}

            Resulting yaml:
            !unknown_tag
            - some content
            """
            key = data.keys()[0]
            if key.startswith("!"):
                value = data[key]
                if type(value) is dict:
                    node = dumper.represent_mapping(key, value)
                elif type(value) is list:
                    node = dumper.represent_sequence(key, value)
                else:
                    node = dumper.represent_scalar(key, value)
            else:
                node = dumper.represent_mapping(u'tag:yaml.org,2002:map', data)
            return node

        yaml.add_representer(dict, representer)
        with self.__get_file("w") as file_obj:
            yaml.dump_all(self.__documents, file_obj,
                          default_flow_style=self.default_flow_style,
                          default_style=self.default_style)

def _folded_unicode_representer(dumper, data: str):
    return dumper.represent_scalar(u'tag:yaml.org,2002:str', data, style='>')

def _literal_unicode_representer(dumper, data: str):
    return dumper.represent_scalar(u'tag:yaml.org,2002:str', data, style='|')

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 dump(self):
        dumpdata = {}
        dumpdata['hosts']       = self.dump_hosts()
        dumpdata['hostgroup']   = self.dump_hostgroups()
        dumpdata['architecture'] = self.dump_arch()
        dumpdata['environment']  = self.dump_env()
        dumpdata['os']           = self.dump_os()
        dumpdata['model']        = self.dump_model()
        dumpdata['media']        = self.dump_media()
        dumpdata['domain']       = self.dump_domain()
        dumpdata['settings']     = self.dump_settings()
        dumpdata['subnet']       = self.dump_subnet()
        dumpdata['smart-proxy']  = self.dump_smartproxy()
        dumpdata['partition-table']  = self.dump_ptable()
        dumpdata['provisioning-template']  = self.dump_provisioningtpl()
        dumpdata['users']        = self.dump_users()
        dumpdata['users']        = self.dump_users()
        dumpdata['auth-source-ldap'] = self.dump_ldaps()
        dumpdata['usergroups'] = self.dump_usergroups()
        dumpdata['roles'] = self.dump_roles()

        # print the result
        fmyml = { 'foreman': dumpdata }

        def str_presenter(dumper, data):
            try:
                dlen = len(data.splitlines())
            except TypeError:
                return dumper.represent_scalar('tag:yaml.org,2002:str', data)
            if (dlen > 1):
                return dumper.represent_scalar('tag:yaml.org,2002:str', data, style='|')

            return dumper.represent_scalar('tag:yaml.org,2002:str', data)

        yaml.add_representer(unicode, str_presenter)
        yaml.add_representer(str, str_presenter)

        yml = yaml.dump(fmyml, allow_unicode=True, default_flow_style=False )
        print( (yml) )

def yaml(self, verbose=False):
        yaml.SafeDumper.add_representer(
            OrderedDict,
            lambda dumper, value: represent_odict(
                dumper, u'tag:yaml.org,2002:map', value
            )
        )

        return yaml.safe_dump(
            self.ordered_dict(verbose=verbose),
            default_flow_style=False,
            indent=2
        ).strip()

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("tag:yaml.org,2002:map", value)

def represent_bool(self, data):
        """Represent bool as 'yes' or 'no' instead of 'true' or 'false'.
        """
        if data:
            value = u'yes'
        else:
            value = u'no'
        return self.represent_scalar('tag:yaml.org,2002:bool', value)

def represent_none(self, data):
        """Represent a None value with nothing instead of 'none'.
        """
        return self.represent_scalar('tag:yaml.org,2002:null', '')

def __init__(self):
        if not toml:
            raise LibraryRequiredError('toml', 'https://pypi.python.org/pypi/toml')

def __init__(self):
        if not yaml:
            raise LibraryRequiredError('PyYAML', 'https://pypi.python.org/pypi/PyYAML')

def load_ordered_config(config_path):
    """
      Loads the configuration in the same order as it's defined in yaml file,
      so that, while saving it in new format, order is maintained
      Args:
            config_path (str): Path to the configuration file
      Returns:
            config(dict): Returns the configurations in the defined ordered
    """

    #  To load data from yaml in ordered dict format
    _mapping_tag = yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG

    def dict_representer(dumper, data):
        return dumper.represent_mapping(_mapping_tag, data.iteritems())

    def dict_constructor(loader, node):
        return collections.OrderedDict(loader.construct_pairs(node))

    yaml.add_representer(collections.OrderedDict, dict_representer)
    yaml.add_constructor(_mapping_tag, dict_constructor)

    #  format the output to print a blank scalar rather than null
    def represent_none(self, _):
        return self.represent_scalar('tag:yaml.org,2002:null', u'')

    yaml.add_representer(type(None), represent_none)

    # read input from home directory for pull requests
    with open(config_path, 'r') as f:
        config = yaml.load(f)
    return config

def multiline_str_representer(dumper, data):
    style = None
    if len(data.splitlines()) > 1:
        style = '|'
    return dumper.represent_scalar('tag:yaml.org,2002:str', data, style=style)

def represent_odict(dumper, data):
        value = list()
        node = yaml.nodes.MappingNode(
            'tag:yaml.org,2002:map', value, flow_style=None )
        if dumper.alias_key is not None:
            dumper.represented_objects[dumper.alias_key] = node
        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))
        node.flow_style = False
        return node

def represent_timedelta(self, delta):
        seconds = delta.seconds
        if not seconds:
            value = '%dd' % delta.days
        elif seconds % 86400 == 0:
            value = '%dd' % (seconds / 86400)
        elif seconds % 3600 == 0:
            value = '%dh' % (seconds / 3600)
        elif seconds % 60 == 0:
            value = '%dm' % (seconds / 60)
        else:
            value = '%ds' % seconds
        return self.represent_scalar('tag:yaml.org,2002:str', value)

def represent_callable(self, obj):
        value = '%s:%s' % (obj.__module__, obj.__name__)
        return self.represent_scalar('tag:yaml.org,2002:str', value)

def represent_decimal(self, data):
        return self.represent_scalar('tag:yaml.org,2002:str', str(data))

def represent_ordered_dict(self, data):
        return self.represent_mapping('tag:yaml.org,2002:map', data.items())

def str_node_representer(dumper, str_node):
    node = yaml.ScalarNode(tag=u'tag:yaml.org,2002:str', value=str(str_node))
    return node

def dict_node_representer(dumper, dict_node):
    items = dict_node.items()
    items = sorted(items, key=lambda x: x[0])
    return dumper.represent_mapping('tag:yaml.org,2002:map', items)

def list_node_representer(dumper, list_node):
    items = sorted(list_node)
    return dumper.represent_mapping('tag:yaml.org,2002:seq', items)

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 __init__(self, *args, **kwargs):
        yaml.SafeLoader.__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 represent_decimal(self, data):
        return self.represent_scalar('tag:yaml.org,2002:str', str(data))

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)