python类YAMLError()的实例源码

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

def create(sysctl_dict, sysctl_file):
    """Creates a sysctl.conf file from a YAML associative array

    :param sysctl_dict: a YAML-formatted string of sysctl options eg "{ 'kernel.max_pid': 1337 }"
    :type sysctl_dict: str
    :param sysctl_file: path to the sysctl file to be saved
    :type sysctl_file: str or unicode
    :returns: None
    """
    try:
        sysctl_dict_parsed = yaml.safe_load(sysctl_dict)
    except yaml.YAMLError:
        log("Error parsing YAML sysctl_dict: {}".format(sysctl_dict),
            level=ERROR)
        return

    with open(sysctl_file, "w") as fd:
        for key, value in sysctl_dict_parsed.items():
            fd.write("{}={}\n".format(key, value))

    log("Updating sysctl_file: %s values: %s" % (sysctl_file, sysctl_dict_parsed),
        level=DEBUG)

    check_call(["sysctl", "-p", sysctl_file])

def load(self):
        # Prepare + load directory.
        super().load()

        # Load the files and parse Yaml.
        parsed_settings = dict()

        try:
            for file_name in self.files:
                file_path = os.path.join(self.directory, file_name)
                with open(file_path, 'r') as file_handle:
                    parsed_settings.update(yaml.load(file_handle))
        except (yaml.YAMLError, yaml.MarkedYAMLError) as e:
            raise ImproperlyConfigured(
                'Your settings file(s) contain invalid YAML syntax! Please fix and restart!, {}'.format(str(e))
            )

        # Loop and set in local settings (+ uppercase keys).
        for key, value in parsed_settings.items():
            self.settings[key.upper()] = value

def create(sysctl_dict, sysctl_file):
    """Creates a sysctl.conf file from a YAML associative array

    :param sysctl_dict: a YAML-formatted string of sysctl options eg "{ 'kernel.max_pid': 1337 }"
    :type sysctl_dict: str
    :param sysctl_file: path to the sysctl file to be saved
    :type sysctl_file: str or unicode
    :returns: None
    """
    try:
        sysctl_dict_parsed = yaml.safe_load(sysctl_dict)
    except yaml.YAMLError:
        log("Error parsing YAML sysctl_dict: {}".format(sysctl_dict),
            level=ERROR)
        return

    with open(sysctl_file, "w") as fd:
        for key, value in sysctl_dict_parsed.items():
            fd.write("{}={}\n".format(key, value))

    log("Updating sysctl_file: %s values: %s" % (sysctl_file, sysctl_dict_parsed),
        level=DEBUG)

    check_call(["sysctl", "-p", sysctl_file])

def create(sysctl_dict, sysctl_file):
    """Creates a sysctl.conf file from a YAML associative array

    :param sysctl_dict: a YAML-formatted string of sysctl options eg "{ 'kernel.max_pid': 1337 }"
    :type sysctl_dict: str
    :param sysctl_file: path to the sysctl file to be saved
    :type sysctl_file: str or unicode
    :returns: None
    """
    try:
        sysctl_dict_parsed = yaml.safe_load(sysctl_dict)
    except yaml.YAMLError:
        log("Error parsing YAML sysctl_dict: {}".format(sysctl_dict),
            level=ERROR)
        return

    with open(sysctl_file, "w") as fd:
        for key, value in sysctl_dict_parsed.items():
            fd.write("{}={}\n".format(key, value))

    log("Updating sysctl_file: %s values: %s" % (sysctl_file, sysctl_dict_parsed),
        level=DEBUG)

    check_call(["sysctl", "-p", sysctl_file])

def main():
    # Read the configuration file
    config_file = os.path.join(rumps.application_support(APP_TITLE),
                               'config.yaml')
    with open(config_file, 'r') as conf:
        try:
            config = yaml.safe_load(conf)
        except yaml.YAMLError as exc:
            print(exc)
            return

    # Setup our CTRL+C signal handler
    signal.signal(signal.SIGINT, _signal_handler)

    # Enable debug mode
    rumps.debug_mode(config['debug'])

    # Startup application
    SlackStatusBarApp(config).run()

def create(sysctl_dict, sysctl_file):
    """Creates a sysctl.conf file from a YAML associative array

    :param sysctl_dict: a YAML-formatted string of sysctl options eg "{ 'kernel.max_pid': 1337 }"
    :type sysctl_dict: str
    :param sysctl_file: path to the sysctl file to be saved
    :type sysctl_file: str or unicode
    :returns: None
    """
    try:
        sysctl_dict_parsed = yaml.safe_load(sysctl_dict)
    except yaml.YAMLError:
        log("Error parsing YAML sysctl_dict: {}".format(sysctl_dict),
            level=ERROR)
        return

    with open(sysctl_file, "w") as fd:
        for key, value in sysctl_dict_parsed.items():
            fd.write("{}={}\n".format(key, value))

    log("Updating sysctl_file: %s values: %s" % (sysctl_file, sysctl_dict_parsed),
        level=DEBUG)

    check_call(["sysctl", "-p", sysctl_file])

def create(sysctl_dict, sysctl_file):
    """Creates a sysctl.conf file from a YAML associative array

    :param sysctl_dict: a YAML-formatted string of sysctl options eg "{ 'kernel.max_pid': 1337 }"
    :type sysctl_dict: str
    :param sysctl_file: path to the sysctl file to be saved
    :type sysctl_file: str or unicode
    :returns: None
    """
    try:
        sysctl_dict_parsed = yaml.safe_load(sysctl_dict)
    except yaml.YAMLError:
        log("Error parsing YAML sysctl_dict: {}".format(sysctl_dict),
            level=ERROR)
        return

    with open(sysctl_file, "w") as fd:
        for key, value in sysctl_dict_parsed.items():
            fd.write("{}={}\n".format(key, value))

    log("Updating sysctl_file: %s values: %s" % (sysctl_file, sysctl_dict_parsed),
        level=DEBUG)

    check_call(["sysctl", "-p", sysctl_file])

def __init__(self):
        self._classification_rules = {}
        self._exclusion_rules = {}
        self._enrichment_rules = {}
        self._routing_rules = {}
        self._config = None

        try:
            # TODO: Absolute path? Where should this live?
            with open('config/klaxer.yml', 'r') as ymlfile:
                self._config = yaml.load(ymlfile)
        except yaml.YAMLError as ye:
            raise ConfigurationError('failed to parse config') from ye

        for section in self._config:
            # Subsequent definitions of the same service will overwrite the
            # previous ones.
            self._build_rules(section)

def create(sysctl_dict, sysctl_file):
    """Creates a sysctl.conf file from a YAML associative array

    :param sysctl_dict: a YAML-formatted string of sysctl options eg "{ 'kernel.max_pid': 1337 }"
    :type sysctl_dict: str
    :param sysctl_file: path to the sysctl file to be saved
    :type sysctl_file: str or unicode
    :returns: None
    """
    try:
        sysctl_dict_parsed = yaml.safe_load(sysctl_dict)
    except yaml.YAMLError:
        log("Error parsing YAML sysctl_dict: {}".format(sysctl_dict),
            level=ERROR)
        return

    with open(sysctl_file, "w") as fd:
        for key, value in sysctl_dict_parsed.items():
            fd.write("{}={}\n".format(key, value))

    log("Updating sysctl_file: %s values: %s" % (sysctl_file, sysctl_dict_parsed),
        level=DEBUG)

    check_call(["sysctl", "-p", sysctl_file])

def create(sysctl_dict, sysctl_file):
    """Creates a sysctl.conf file from a YAML associative array

    :param sysctl_dict: a YAML-formatted string of sysctl options eg "{ 'kernel.max_pid': 1337 }"
    :type sysctl_dict: str
    :param sysctl_file: path to the sysctl file to be saved
    :type sysctl_file: str or unicode
    :returns: None
    """
    try:
        sysctl_dict_parsed = yaml.safe_load(sysctl_dict)
    except yaml.YAMLError:
        log("Error parsing YAML sysctl_dict: {}".format(sysctl_dict),
            level=ERROR)
        return

    with open(sysctl_file, "w") as fd:
        for key, value in sysctl_dict_parsed.items():
            fd.write("{}={}\n".format(key, value))

    log("Updating sysctl_file: %s values: %s" % (sysctl_file, sysctl_dict_parsed),
        level=DEBUG)

    check_call(["sysctl", "-p", sysctl_file])

def create(sysctl_dict, sysctl_file):
    """Creates a sysctl.conf file from a YAML associative array

    :param sysctl_dict: a YAML-formatted string of sysctl options eg "{ 'kernel.max_pid': 1337 }"
    :type sysctl_dict: str
    :param sysctl_file: path to the sysctl file to be saved
    :type sysctl_file: str or unicode
    :returns: None
    """
    try:
        sysctl_dict_parsed = yaml.safe_load(sysctl_dict)
    except yaml.YAMLError:
        log("Error parsing YAML sysctl_dict: {}".format(sysctl_dict),
            level=ERROR)
        return

    with open(sysctl_file, "w") as fd:
        for key, value in sysctl_dict_parsed.items():
            fd.write("{}={}\n".format(key, value))

    log("Updating sysctl_file: %s values: %s" % (sysctl_file, sysctl_dict_parsed),
        level=DEBUG)

    check_call(["sysctl", "-p", sysctl_file])

def create(sysctl_dict, sysctl_file):
    """Creates a sysctl.conf file from a YAML associative array

    :param sysctl_dict: a YAML-formatted string of sysctl options eg "{ 'kernel.max_pid': 1337 }"
    :type sysctl_dict: str
    :param sysctl_file: path to the sysctl file to be saved
    :type sysctl_file: str or unicode
    :returns: None
    """
    try:
        sysctl_dict_parsed = yaml.safe_load(sysctl_dict)
    except yaml.YAMLError:
        log("Error parsing YAML sysctl_dict: {}".format(sysctl_dict),
            level=ERROR)
        return

    with open(sysctl_file, "w") as fd:
        for key, value in sysctl_dict_parsed.items():
            fd.write("{}={}\n".format(key, value))

    log("Updating sysctl_file: %s values: %s" % (sysctl_file, sysctl_dict_parsed),
        level=DEBUG)

    check_call(["sysctl", "-p", sysctl_file])

def create(sysctl_dict, sysctl_file):
    """Creates a sysctl.conf file from a YAML associative array

    :param sysctl_dict: a YAML-formatted string of sysctl options eg "{ 'kernel.max_pid': 1337 }"
    :type sysctl_dict: str
    :param sysctl_file: path to the sysctl file to be saved
    :type sysctl_file: str or unicode
    :returns: None
    """
    try:
        sysctl_dict_parsed = yaml.safe_load(sysctl_dict)
    except yaml.YAMLError:
        log("Error parsing YAML sysctl_dict: {}".format(sysctl_dict),
            level=ERROR)
        return

    with open(sysctl_file, "w") as fd:
        for key, value in sysctl_dict_parsed.items():
            fd.write("{}={}\n".format(key, value))

    log("Updating sysctl_file: %s values: %s" % (sysctl_file, sysctl_dict_parsed),
        level=DEBUG)

    check_call(["sysctl", "-p", sysctl_file])

def read_database(self):
        """
            Read .yml database files provided by 2factorauth guys!
        """
        db_dir = path.join(env.get("DATA_DIR"), "applications") + "/data/*.yml"
        logging.debug("Database folder is {0}".format(db_dir))
        db_files = glob(db_dir)
        logging.debug("Reading database files started")
        for db_file in db_files:
            logging.debug("Reading database file {0}".format(db_file))
            with open(db_file, 'r') as data:
                try:
                    websites = yaml.load(data)["websites"]
                    for app in websites:
                        if self.is_valid_app(app):
                            self.db.append(app)
                except yaml.YAMLError as error:
                    logging.error("Error loading yml file {0} : {1}".format(
                        db_file, str(error)))
                except TypeError:
                    logging.error("Not a valid yml file {0}".format(db_file))
        logging.debug("Reading database files finished")

def get_ha_proxy_address(ctx):
    client = index_client.IndexClient(ctx.meta['index-file'])
    manifest = client.find_by_release('elastic-runtime')
    settings = ctx.meta['settings']

    username = settings["username"]
    manifest_path = os.path.join("/home", username, 'manifests', manifest['file'])

    with open(manifest_path, 'r') as stream:
        content = stream.read()

        try:
            doc = yaml.load(content)
        except yaml.YAMLError as exc:
            print(exc)
            return None

    job = filter(lambda job: job['name'] == 'haproxy', doc['jobs'])[0]
    default_net = filter(lambda net: net['name'] == 'default', job['networks'])[0]
    return default_net['static_ips'][0]

def _load_config(self, config_files, profile):
        """Initialize client settings from config."""
        for filename in config_files:
            try:
                with open(os.path.expanduser(filename), 'r') as f:
                    config = yaml.safe_load(f.read())
                    return config.get(profile, {})
            except IOError:
                logging.debug('{0} config file not found.'.format(filename))
                pass
            except yaml.YAMLError as e:
                msg = 'Failed to parse {0}: {1}'.format(filename, e)
                logging.error(msg)
                raise ClientConfigurationError(msg)
        # No file found
        return {}

def _load(self):
            """Load config file into memory."""
            try:
                with open(self.file, 'r') as file:
                    self._data = yaml.load(file)
            except IOError:
                # no config file
                pass
            except yaml.YAMLError as err:
                raise Exception(
                    'Could not parse corrupt config file: %s\n'
                    'Try running "rm %s"' % (
                        str(err),
                        self.file,
                    )
                )
            # set defaults
            for key, value in Config.defaults.items():
                if key not in self._data:
                    self._data[key] = value

def load_yaml(fname: str) -> Union[List, Dict]:
    """Load a YAML file."""
    try:
        with open(fname, encoding='utf-8') as conf_file:
            # If configuration file is empty YAML returns None
            # We convert that to an empty dict
            return yaml.load(conf_file, Loader=SafeLineLoader) or {}
    except yaml.YAMLError as exc:
        logger.error(exc)
        raise ScarlettError(exc)
    except UnicodeDecodeError as exc:
        logger.error('Unable to read file %s: %s', fname, exc)
        raise ScarlettError(exc)


# def clear_secret_cache() -> None:
#     """Clear the secret cache.
#
#     Async friendly.
#     """
#     __SECRET_CACHE.clear()

def load(file):
    try:
        config_file = open(file, 'r')
    except IOError as e:
        sys.exit(
            'Error while opening required file: "{}"\n'
            '{}'
                .format(file, e)
        )

    with config_file:
        try:
            config = yaml.load(config_file)
        except yaml.YAMLError as e:
            sys.exit(
                'Error while parsing file: "{}"\n'
                '{}'
                    .format(file, e)
            )

    return config

def load_configuration(conf):
    """Load the swiftbackmeup configuration file."""

    file_path = check_configuration_file_existence(conf.get('file_path'))

    try:
        file_path_content = open(file_path, 'r').read()
    except IOError as exc:
        raise exceptions.ConfigurationExceptions(exc)

    try:
        conf = yaml.load(file_path_content)
    except yaml.YAMLError as exc:
        raise exceptions.ConfigurationExceptions(exc)

    return conf

def on_post(self, req, resp, revision_id, tag=None):
        """Creates a revision tag."""
        body = req.stream.read(req.content_length or 0)

        try:
            tag_data = yaml.safe_load(body)
        except yaml.YAMLError as e:
            error_msg = ("Could not parse the request body into YAML data. "
                         "Details: %s." % e)
            LOG.error(error_msg)
            raise falcon.HTTPBadRequest(description=e)

        try:
            resp_tag = db_api.revision_tag_create(revision_id, tag, tag_data)
        except errors.RevisionNotFound as e:
            raise falcon.HTTPNotFound(description=e.format_message())
        except errors.RevisionTagBadFormat as e:
            raise falcon.HTTPBadRequest(description=e.format_message())

        resp_body = revision_tag_view.ViewBuilder().show(resp_tag)
        resp.status = falcon.HTTP_201
        resp.body = resp_body

def create(sysctl_dict, sysctl_file):
    """Creates a sysctl.conf file from a YAML associative array

    :param sysctl_dict: a YAML-formatted string of sysctl options eg "{ 'kernel.max_pid': 1337 }"
    :type sysctl_dict: str
    :param sysctl_file: path to the sysctl file to be saved
    :type sysctl_file: str or unicode
    :returns: None
    """
    try:
        sysctl_dict_parsed = yaml.safe_load(sysctl_dict)
    except yaml.YAMLError:
        log("Error parsing YAML sysctl_dict: {}".format(sysctl_dict),
            level=ERROR)
        return

    with open(sysctl_file, "w") as fd:
        for key, value in sysctl_dict_parsed.items():
            fd.write("{}={}\n".format(key, value))

    log("Updating sysctl_file: %s values: %s" % (sysctl_file, sysctl_dict_parsed),
        level=DEBUG)

    check_call(["sysctl", "-p", sysctl_file])

def create(sysctl_dict, sysctl_file):
    """Creates a sysctl.conf file from a YAML associative array

    :param sysctl_dict: a YAML-formatted string of sysctl options eg "{ 'kernel.max_pid': 1337 }"
    :type sysctl_dict: str
    :param sysctl_file: path to the sysctl file to be saved
    :type sysctl_file: str or unicode
    :returns: None
    """
    try:
        sysctl_dict_parsed = yaml.safe_load(sysctl_dict)
    except yaml.YAMLError:
        log("Error parsing YAML sysctl_dict: {}".format(sysctl_dict),
            level=ERROR)
        return

    with open(sysctl_file, "w") as fd:
        for key, value in sysctl_dict_parsed.items():
            fd.write("{}={}\n".format(key, value))

    log("Updating sysctl_file: %s values: %s" % (sysctl_file, sysctl_dict_parsed),
        level=DEBUG)

    check_call(["sysctl", "-p", sysctl_file])

def _load_config():
    # Fills the global CONFIG dictionary using default and custom config
    # Returns an error if the custom config is invalid
    global CONFIG
    try:
        cfg = _load_default_config()
        custom_cfg = _load_custom_config()
        if custom_cfg:
            CONFIG = _merge(cfg, custom_cfg)
        else:
            CONFIG = cfg
    except yaml.YAMLError as exc:
        # Try to point to the line that threw an error
        if hasattr(exc, 'problem_mark'):
            mark = exc.problem_mark
            return 'Error in YAML at position: ({}:{})'.format(mark.line + 1,
                                                               mark.column + 1)

def req_yaml(self, req):
        if req.content_length is None or req.content_length == 0:
            return None

        raw_body = req.stream.read(req.content_length or 0)

        if raw_body is None:
            return None

        try:
            return yaml.safe_load_all(raw_body.decode('utf-8'))
        except yaml.YAMLError as jex:
            self.error(
                req.context,
                "Invalid YAML in request: \n%s" % raw_body.decode('utf-8'))
            raise Exception(
                "%s: Invalid YAML in body: %s" % (req.path, jex))

def create(sysctl_dict, sysctl_file):
    """Creates a sysctl.conf file from a YAML associative array

    :param sysctl_dict: a YAML-formatted string of sysctl options eg "{ 'kernel.max_pid': 1337 }"
    :type sysctl_dict: str
    :param sysctl_file: path to the sysctl file to be saved
    :type sysctl_file: str or unicode
    :returns: None
    """
    try:
        sysctl_dict_parsed = yaml.safe_load(sysctl_dict)
    except yaml.YAMLError:
        log("Error parsing YAML sysctl_dict: {}".format(sysctl_dict),
            level=ERROR)
        return

    with open(sysctl_file, "w") as fd:
        for key, value in sysctl_dict_parsed.items():
            fd.write("{}={}\n".format(key, value))

    log("Updating sysctl_file: %s values: %s" % (sysctl_file, sysctl_dict_parsed),
        level=DEBUG)

    check_call(["sysctl", "-p", sysctl_file])

def create(sysctl_dict, sysctl_file):
    """Creates a sysctl.conf file from a YAML associative array

    :param sysctl_dict: a YAML-formatted string of sysctl options eg "{ 'kernel.max_pid': 1337 }"
    :type sysctl_dict: str
    :param sysctl_file: path to the sysctl file to be saved
    :type sysctl_file: str or unicode
    :returns: None
    """
    try:
        sysctl_dict_parsed = yaml.safe_load(sysctl_dict)
    except yaml.YAMLError:
        log("Error parsing YAML sysctl_dict: {}".format(sysctl_dict),
            level=ERROR)
        return

    with open(sysctl_file, "w") as fd:
        for key, value in sysctl_dict_parsed.items():
            fd.write("{}={}\n".format(key, value))

    log("Updating sysctl_file: %s values: %s" % (sysctl_file, sysctl_dict_parsed),
        level=DEBUG)

    check_call(["sysctl", "-p", sysctl_file])

def parse_infile(file):
    """ Parse data structure from file into dictionary for component use """
    with open(file, 'r') as data_file:
        try:
            data = json.load(data_file)
            logging.debug("Data parsed from file {}: {}".format(file, data))
            return data
        except ValueError as json_error:
            pass

        data_file.seek(0)

        try:
            data = yaml.load(data_file)
            logging.debug("Data parsed from file {}: {}".format(file, data))
            return data
        except yaml.YAMLError as yaml_error:
            pass

    logging.error("Unable to parse in file. {} {} ".format(json_error, yaml_error))

def parse_yaml(stream, overrides={}):
    import yaml
    import sys
    from autocmake.interpolate import interpolate

    try:
        config = yaml.load(stream, yaml.SafeLoader)
    except yaml.YAMLError as exc:
        print(exc)
        sys.exit(-1)

    for k in config:
        if k in overrides:
            config[k] = overrides[k]

    config = interpolate(config, config)
    return config