def main():
parser = setup_parser()
options = parser.parse_args()
log_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
logging.basicConfig(level=logging.DEBUG, format=log_format)
logging.getLogger('botocore').setLevel(logging.WARNING)
with open(options.config) as fh:
config = yaml.load(fh.read(), Loader=yaml.SafeLoader)
jsonschema.validate(config, CONFIG_SCHEMA)
setup_defaults(config)
tester = MailerTester(
options.MESSAGE_FILE, config, msg_plain=options.plain,
json_dump_file=options.json_dump_file
)
tester.run(options.dry_run, options.print_only)
python类SafeLoader()的实例源码
def _ordered_load(stream, Loader=yaml.Loader, object_pairs_hook=OrderedDict):
"""
Ordered yaml loader
Use this instead ot yaml.loader/yaml.saveloader to get an Ordereddict
:param stream: stream to read from
:param Loader: yaml-loader to use
:object_pairs_hook: ...
:return: OrderedDict structure
"""
# usage example: 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 load(stream, Loader=yaml.SafeLoader, object_pairs_hook=OrderedDict):
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 save(data, stream=None, Dumper=yaml.SafeDumper,
default_flow_style=False,
encoding='utf-8',
**kwds):
class OrderedDumper(Dumper):
pass
def _dict_representer(dumper, data):
return dumper.represent_mapping(
yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
data.items())
OrderedDumper.add_representer(OrderedDict, _dict_representer)
OrderedDumper.add_representer(np.float64,
lambda dumper, data: dumper.represent_float(float(data)))
OrderedDumper.add_representer(complex,
lambda dumper, data: dumper.represent_str(str(data)))
OrderedDumper.add_representer(np.complex128,
lambda dumper, data: dumper.represent_str(str(data)))
OrderedDumper.add_representer(np.ndarray,
lambda dumper, data: dumper.represent_list(list(data)))
# I added the following two lines to make pyrpl compatible with pyinstruments. In principle they can be erased
if isinstance(data, dict) and not isinstance(data, OrderedDict):
data = OrderedDict(data)
return yaml.dump(data,
stream=stream,
Dumper=OrderedDumper,
default_flow_style=default_flow_style,
encoding=encoding,
**kwds)
# usage example:
# load(stream, yaml.SafeLoader)
# save(data, stream=f, Dumper=yaml.SafeDumper)
def get_and_validate_mailer_config(args):
with open(args.config) as fh:
config = yaml.load(fh.read(), Loader=yaml.SafeLoader)
jsonschema.validate(config, CONFIG_SCHEMA)
utils.setup_defaults(config)
return config
def yaml_load_fromstring(string, ordered=False):
"""
Load contents of a string into an dict/OrderedDict structure. The string has to be valid yaml
:param string: name of the yaml file to load
:type string: str
:param ordered: load to an OrderedDict? Default=False
:type ordered: bool
:return: configuration data loaded from the file (or None if an error occured)
:rtype: Dict | OrderedDict | None
"""
dict_type = 'dict'
if ordered:
dict_type = 'OrderedDict'
logger.info("Loading '{}' to '{}'".format(string, dict_type))
y = None
estr = ''
try:
sdata = string
# sdata = sdata.replace('\n', '\n\n')
if ordered:
y = _ordered_load(sdata, yaml.SafeLoader)
else:
y = yaml.load(sdata, yaml.SafeLoader)
except Exception as e:
estr = str(e)
if "found character '\\t'" in estr:
estr = estr[estr.find('line'):]
estr = 'TABs are not allowed in YAML files, use spaces for indentation instead!\nError in ' + estr
if ("while scanning a simple key" in estr) and ("could not found expected ':'" in estr):
estr = estr[estr.find('column'):estr.find('could not')]
estr = 'The colon (:) following a key has to be followed by a space. The space is missing!\nError in ' + estr
return y, estr
def load(filename):
"""
Loads a config file
@param filename str: filename of config file
Returns OrderedDict
"""
with open(filename, 'r') as f:
return ordered_load(f, yaml.SafeLoader)
def yaml_load(filename, ordered=False, ignore_notfound=False):
"""
Load contents of a configuration file into an dict/OrderedDict structure. The configuration file has to be a valid yaml file
:param filename: name of the yaml file to load
:type filename: str
:param ordered: load to an OrderedDict? Default=False
:type ordered: bool
:return: configuration data loaded from the file (or None if an error occured)
:rtype: Dict | OrderedDict | None
"""
dict_type = 'dict'
if ordered:
dict_type = 'OrderedDict'
logger.info("Loading '{}' to '{}'".format(filename, dict_type))
y = None
try:
with open(filename, 'r') as stream:
sdata = stream.read()
sdata = sdata.replace('\n', '\n\n')
if ordered:
y = _ordered_load(sdata, yaml.SafeLoader)
else:
y = yaml.load(sdata, yaml.SafeLoader)
except Exception as e:
estr = str(e)
if "found character '\\t'" in estr:
estr = estr[estr.find('line'):]
estr = 'TABs are not allowed in YAML files, use spaces for indentation instead!\nError in ' + estr
if ("while scanning a simple key" in estr) and ("could not found expected ':'" in estr):
estr = estr[estr.find('column'):estr.find('could not')]
estr = 'The colon (:) following a key has to be followed by a space. The space is missing!\nError in ' + estr
if "[Errno 2]" in estr:
if not ignore_notfound:
logger.warning("YAML-file not found: {}".format(filename))
else:
logger.error("YAML-file load error in {}: \n{}".format(filename, estr))
return y
def readConfig(configType):
if configType in configurationType:
# The earliest config files are given the _most_ precedence.
# ie. A value in the config in the local directory will override the same variable
# defined in the config in the package base directory.
# For more on pkg_resources, see: https://stackoverflow.com/a/5601839
fileList = [
# Config file in the local directory where it is run
"config.yaml",
# Config in the home directory
# Ensures that we have "WebApp" here.
os.path.expandvars("~/.overwatch{0}").format(configType.name[0].upper() + configType.name[1:]),
# Config type specific directory in the package (ex: "processing")
# TODO: There is a problem when loading the shared configuration with the processing configuration
# because the shared configuration can have options which are defined in the web app config
# and therefore undefined when the web app config is not loaded!
# To resolve it temporarily, both configuration files will be included
pkg_resources.resource_filename("overwatch.webApp", "config.yaml"),
pkg_resources.resource_filename("overwatch.processing", "config.yaml"),
pkg_resources.resource_filename("overwatch.receiver", "config.yaml"),
pkg_resources.resource_filename("overwatch.api", "config.yaml"),
# Below is the line that should be used when the above issue is resolved
#pkg_resources.resource_filename("overwatch.{0}".format(configType.name), "config.yaml"),
# Shared config in the package base
pkg_resources.resource_filename("overwatch.base", "config.yaml")
]
else:
# Cannot just be the logger because the logger many not yet be initialized
print("CRITICAL: Unrecognized configuration type {0}!".format(configType.name))
logger.critical("Unrecognized configuration type {0}!".format(configType.name))
sys.exit(1)
# Suppressed for cleaning up start up messages
#logger.debug("Config filenames: {0}".format(fileList))
(configs, filesRead) = readConfigFiles(fileList)
# Suppressed for cleaning up start up messages
#logger.debug("Read config files: {0}".format(filesRead))
# Merge the configurations together
# List is reversed so the earlier listed config will always override settings from lower listed files
configs = "\n".join(reversed(configs))
#print("configs: {0}".format(configs))
# Handle warnings
# See: https://stackoverflow.com/a/40376576
with warnings.catch_warnings():
warnings.simplefilter("ignore")
globalConfig = yaml.load(configs, Loader=yaml.SafeLoader)
return (globalConfig, filesRead)
