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

def parse_configs():
    """Parses config files out of config/ directory.

    Returns:
        Array of MergeBotConfig objects.
    """
    configs = []
    yaml.add_constructor(u'!MergeBotConfig', mergebotconfig_constructor)
    l.info('Parsing Config Files')
    for filename in glob.iglob('config/*.yaml'):
        with open(filename) as cfg:
            try:
                l.info('Opening {}'.format(filename))
                config = yaml.load(cfg)
                l.info('{} Successfully Read'.format(filename))
                configs.append(config)
            except yaml.YAMLError as exc:
                l.fatal(
                    'Error parsing file {filename}: {exc}. Please fix and try '
                    'again.'.format(filename=filename, exc=exc))
    return configs

def parse_config(filename):
    """ parses the YAML config file and expands any environment variables """

    pattern = re.compile(r'^\<%= ENV\[\'(.*)\'\] %\>(.*)$')
    yaml.add_implicit_resolver("!pathex", pattern)

    def pathex_constructor(loader, node):
        value = loader.construct_scalar(node)
        envVar, remainingPath = pattern.match(value).groups()
        return os.environ[envVar] + remainingPath

    yaml.add_constructor('!pathex', pathex_constructor)

    with open(filename) as f:
        return(
            yaml.load(f)
        )

def setup_yaml_parser():
    var = re.compile(r".*\$\{.*\}.*", re.VERBOSE)
    yaml.add_constructor('!env_var', _env_var_constructor)
    yaml.add_implicit_resolver('!env_var', var)

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 __init__(self, logger):
        """
        Initialize the class
        :param logger: Logging
        """
        super(YamlConfiguration, self).__init__()
        yaml.add_constructor('!include', self.__yaml_include)
        self.__logger = logger

def parse(u):
    """
    Parse the contents of a spec test file, and return a dict.

    Arguments:

      u: a unicode string.

    """
    # TODO: find a cleaner mechanism for choosing between the two.
    if yaml is None:
        # Then use json.

        # The only way to get the simplejson module to return unicode strings
        # is to pass it unicode.  See, for example--
        #
        #   http://code.google.com/p/simplejson/issues/detail?id=40
        #
        # and the documentation of simplejson.loads():
        #
        #   "If s is a str then decoded JSON strings that contain only ASCII
        #    characters may be parsed as str for performance and memory reasons.
        #    If your code expects only unicode the appropriate solution is
        #    decode s to unicode prior to calling loads."
        #
        return json.loads(u)
    # Otherwise, yaml.

    def code_constructor(loader, node):
        value = loader.construct_mapping(node)
        return eval(value['python'], {})

    yaml.add_constructor(u'!code', code_constructor)
    return yaml.load(u)

def __init__(self, **options):
        super(YamlReaderWriter, self).__init__(**options)

        try:
            import yaml
            import yaml.resolver
        except ImportError:
            raise RuntimeError('To use YAML, please install PyYAML first')

        #
        # The code to preserve order of items is taken from here:
        # https://stackoverflow.com/a/21048064/38611
        #

        _mapping_tag = yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG

        def dict_representer(dumper, data):
            return dumper.represent_dict(data.items())

        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)

        self.yaml = yaml

        self.default_dump_options = {
            'indent': 2,
            'default_flow_style': False,
        }

def config_file_as_dict(**kwargs):
    cfgfile = kwargs["cfgfile"]
    cfgfile_contents = templated_file_contents(kwargs, kwargs["cfgfile"])
#    print (cfgfile_contents)
    cfg_data = {}
    if cfgfile.endswith(".json"):
        cfgdata = json.loads(cfgfile_contents)
    elif cfgfile.endswith(".toml"):
        cfgdata = toml.loads(cfgfile_contents)
    elif cfgfile.endswith(".yaml"):
        yaml.add_constructor('!join', join) # http://stackoverflow.com/questions/5484016/how-can-i-do-string-concatenation-or-string-replacement-in-yaml
        cfgdata = yaml.load(cfgfile_contents)
    else:
        raise ValueError("Invalid config file format")
    return merge_two_dicts(kwargs, cfgdata)

def setup_yaml():
    _mapping_tag = yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG

    def dict_representer(dumper, data):
        return dumper.represent_dict(data.items())

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

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

def parse_config(config_file):
    """Parse the YAML config"""
    pattern = re.compile(r'^\<%= ENV\[\'(.*)\'\] %\>(.*)$')
    yaml.add_implicit_resolver("!env", pattern)

    def env_constructor(loader, node):
        """Constructor for environment variables"""
        value = loader.construct_scalar(node)
        env_var, remaining_path = pattern.match(value).groups()
        return os.environ[env_var] + remaining_path

    yaml.add_constructor('!env', env_constructor)
    with open(config_file) as config:
        return yaml.load(config)

def ordered_load(stream, Loader=yaml.Loader, object_pairs_hook=OrderedDict):
    # ordered_load(stream, yaml.SafeLoader)
    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 setup_yaml():
    _mapping_tag = yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG
    yaml.add_representer(collections.OrderedDict, dict_representer)
    yaml.add_representer(os._Environ, dict_representer)
    yaml.add_constructor(_mapping_tag, dict_constructor)

def main():
    """Main application entry point."""
    if len(sys.argv) != 2:
        print "Usage: yamlcalc <input-file>"
        return

    mapping_tag = yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG
    sequence_tag = yaml.resolver.BaseResolver.DEFAULT_SEQUENCE_TAG

    yaml.add_constructor(mapping_tag, dict_constructor)
    yaml.add_constructor(sequence_tag, list_constructor)
    yaml.add_representer(CalcDict, dict_representer)
    yaml.add_representer(CalcList, list_representer)

    try:
        with open(sys.argv[1]) as infp:
            top = yaml.load(infp)

            if not isinstance(top, CalcDict):
                type_name = type(top).__name__
                err("Top level element should be dict not {0}".format(type_name))

            defs = {}
            defs_str = top.get("DEFS", "")

            try:
                exec defs_str in defs
            except Exception as exc:
                err("Error executing DEFS: {0}".format(exc))

            CalcContainer.set_top(defs, top)

            view = top.get("VIEW", {})
            writer_type = view.get("type", "raw")
            writer_func_name = "_".join(writer_type.split("-"))

            try:
                write = globals()["write_" + writer_func_name]
            except KeyError:
                err("Error unsupporter writer: {0}".format(writer_type))

            outdir, ext = os.path.splitext(sys.argv[1])
            if os.path.exists(outdir):
                if not os.path.isdir(outdir):
                    err("Path exists but is not a directory: {0}".format(outdir))
            else:
                try:
                    os.mkdir(outdir)
                except OSError as exc:
                    err("Error create directory: {0}".format(outdir))

            write(view, top, outdir)
    except IOError as exc:
        err("Error opening file: {0}".format(exc))
    except yaml.YAMLError as exc:
        err("Error parsing input: {0}".format(exc))

def parse(c, relative_path):
    """
    Parse RAML file

    :param c: file content
    :type c: str
    :return:
    """

    # Read RAML header
    first_line, c = c.split('\n', 1)
    raml_version = _validate_raml_header(first_line)

    #includes = None
    # Adding yaml !include support before calling the load function
    #try:
    #    includes = yaml.add_constructor("!include", yaml_include)
    #except:
    #    pass

    context = ParseContext(yaml.load(c), relative_path)

    root = RamlRoot(raml_version=raml_version)
    root.title = context.get_string_property('title', True)

    schemas = context.get_string_property('schemas')
    root.schemas = schemas

    root.baseUri = context.get_string_property('baseUri')
    root.version = context.get_string_property('version')
    root.mediaType = context.get_string_property('mediaType')

    root.documentation = context.get_property_with_schema('documentation', RamlRoot.documentation)
    root.traits = parse_traits(context, RamlRoot.traits.field_name)
    root.resourceTypes = parse_resource_type(context)

    resources = OrderedDict()
    for property_name in context.__iter__():
        if property_name.startswith("/"):
            resources[property_name] = parse_resource(context, property_name, root)

    if resources > 0:
        root.resources = resources

    return root

def load_config_file(self, config_paths):
        """Load a yaml config file from path."""
        config_path = ""
        for possible_path in config_paths:
            if not os.path.isfile(possible_path):
                _LOGGER.debug("Config file %s not found.", possible_path)
            else:
                config_path = possible_path
                break

        if not config_path:
            _LOGGER.info("No configuration files found.")
            config_path = self.create_default_config(DEFAULT_CONFIG_PATH)

        env_var_pattern = re.compile(r'^\$([A-Z_]*)$')
        yaml.add_implicit_resolver("!envvar", env_var_pattern)

        def envvar_constructor(loader, node):
            """Yaml parser for env vars."""
            value = loader.construct_scalar(node)
            [env_var] = env_var_pattern.match(value).groups()
            return os.environ[env_var]

        def include_constructor(loader, node):
            """Add a yaml file to be loaded inside another."""
            main_yaml_path = os.path.split(stream.name)[0]
            included_yaml = os.path.join(main_yaml_path,
                                         loader.construct_scalar(node))

            with open(included_yaml, 'r') as included:
                return yaml.load(included)

        yaml.add_constructor('!envvar', envvar_constructor)
        yaml.add_constructor('!include', include_constructor)

        try:
            with open(config_path, 'r') as stream:
                _LOGGER.info("Loaded config from %s.", config_path)
                return yaml.load(stream)
        except yaml.YAMLError as error:
            self.opsdroid.critical(error, 1)
        except FileNotFoundError as error:
            self.opsdroid.critical(str(error), 1)

def parse_config(config_path):
    """Checks if secrets need to be fetched from the environment"""

    temp_path = "%s.tmp" % config_path

    # Template yaml file
    with open(config_path, 'r') as stream:
        try:
            temp = template_yaml(yaml.load(stream))
        except yaml.YAMLError as exc:
            raise ValueError("Invalid config passed to the program: %s" % exc)

    # Dump templated file back to file-system
    with open(temp_path, 'w') as outfile:
        yaml.safe_dump(temp, outfile, default_flow_style=False)

    # Add environment resolver
    pattern_env = re.compile(r'^(.*)\<%= ENV\[\'(.*)\'\] %\>(.*)$')
    yaml.add_implicit_resolver("!pathex", pattern_env)

    # Add command resolver
    pattern_cmd = re.compile(r'^(.*)\<%= CMD\[\'(.*)\'\] %\>(.*)$')
    yaml.add_implicit_resolver("!pathcmd", pattern_cmd)

    # Add function resolver
    pattern_fun = re.compile(r'^(.*)\<%= FUNC\[\'(.*)\((.*)\)\'\] %\>(.*)$')
    yaml.add_implicit_resolver("!func", pattern_fun)

    def pathex_constructor(loader, node):
        """Processes environment variables found in the YAML"""
        value = loader.construct_scalar(node)
        before_path, env_var, remaining_path = pattern_env.match(value).groups()
        return before_path + os.environ[env_var] + remaining_path

    def pathcmd_constructor(loader, node):
        """Processes command variables found in the YAML"""
        value = loader.construct_scalar(node)
        before_path, cmd_var, remaining_path = pattern_cmd.match(value).groups()
        retval = output_value(execute(cmd_var, None, True, True), True)
        return before_path + retval.decode("utf-8") + remaining_path

    def fun_constructor(loader, node):
        """Processes embedded functions found in the YAML"""
        value = loader.construct_scalar(node)
        before_path, fun, arg, remaining_path = pattern_fun.match(value).groups()
        retval = PREDEFINED_FUNCTIONS[fun](arg)
        return before_path + retval.decode("utf-8") + remaining_path

    yaml.add_constructor("!pathex", pathex_constructor)
    yaml.add_constructor("!pathcmd", pathcmd_constructor)
    yaml.add_constructor("!func", fun_constructor)

    with open(temp_path, 'r') as stream:
        try:
            return yaml.load(stream)
        except yaml.YAMLError as exc:
            raise ValueError("Invalid config passed to the program: %s" % exc)