我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用yaml.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 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 __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 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 _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 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
def get_json_from_file(self): file_name = os.path.join(self.path, self.args['file']) try: with open(file_name, 'r', encoding='utf-8') as f: body = json.load(f) return body except FileNotFoundError: self.parser.error('no such file: {}'.format( file_name )) except json.decoder.JSONDecodeError as e: pass try: with open(file_name, 'r', encoding='utf-8') as f: body = yaml.load(f) return body except FileNotFoundError: self.parser.error('no such file: {}'.format( file_name )) except yaml.YAMLError as e: self.parser.error('bad json or yaml: {}'.format( e ))
def post(self, **params): user = self.get_current_user() data = params.get('data') if data is None: raise InsufficientData('No "data" provided') try: parsed_data = list(yaml.safe_load_all(data)) except yaml.YAMLError as e: raise PredefinedAppExc.UnparseableTemplate( 'Incorrect yaml, parsing failed: "{0}"'.format(str(e))) try: res = start_pod_from_yaml(parsed_data, user=user) except APIError as e: # pass as is raise except Exception as e: raise PredefinedAppExc.InternalPredefinedAppError( details={'message': str(e)}) send_event_to_user('pod:change', res, user.id) return res
def _load_status_file(self, fpath): global HTTP_YAML_CACHE if fpath in HTTP_YAML_CACHE: return HTTP_YAML_CACHE[fpath] # self.app.info("Fpath: %s" % fpath) try: with open(fpath, 'r') as stream: lookup = yaml.safe_load(stream) except IOError: self.app.warn( "Parameters file %s not found" % fpath, (self.env.docname, None)) return except yaml.YAMLError as exc: self.app.warn(exc) raise HTTP_YAML_CACHE[fpath] = lookup return lookup