Python ruamel.yaml 模块，SafeLoader() 实例源码

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)

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)