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

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 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 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)