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)
)
python类add_implicit_resolver()的实例源码
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)