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