def generate_cwl_documentation(_):
cur_dir = os.path.abspath(os.path.dirname(__file__))
# find all cwl files
with WorkflowGenerator() as wf:
cwl_files = [step.run for step in wf.steps_library.steps.values()]
# sort alphabetically
cwl_files.sort()
tools_file = os.path.join(cur_dir, 'tools.rst')
tool_template = '\n{}\n{}\n\n{}\n'
with codecs.open(tools_file, 'wb', encoding='utf-8') as f:
f.write('Tools\n=====\n')
f.write('\n``nlppln`` contains the following tools:\n')
for cwl in cwl_files:
tool_name = os.path.basename(cwl)
plusses = '+'*len(tool_name)
with codecs.open(cwl) as c:
try:
cwl_yaml = yaml.load(c, Loader=yaml.RoundTripLoader)
doc = cwl_yaml.get('doc', 'No documentation')
f.write(tool_template.format(tool_name, plusses, doc))
except yaml.YAMLError:
pass
python类YAMLError()的实例源码
def save(self, path: str):
cfg = self._dict_to_orderdict(self.cfg_dict)
with open(path[:-3] + 'yaml', 'w') as cfg_file:
try:
yaml3ed.dump(cfg, cfg_file, explicit_start=True, explicit_end=True,
default_flow_style=False, allow_unicode=True, version=(1, 2),
indent=2)
except yaml3ed.YAMLError as exc:
print(exc)
def load(self, path_cfg: str):
with open(path_cfg, 'r') as stream:
try:
self.cfg_dict = yaml3ed.safe_load(stream)
except yaml3ed.YAMLError as exc:
print(exc)
self.check()
self.zbx = ZabbixAgent(self.cfg_dict['zabbix']['url'], self.cfg_dict['zabbix']['login'],
self.cfg_dict['zabbix']['password'])
log.debug('Config loaded')
def save(self, path: str):
cfg = self._dict_to_orderdict(self.map_config)
with open(path + '/' + self.map_data['name'] + '.yaml', 'w') as cfg_file:
try:
yaml3ed.dump(cfg, cfg_file, explicit_start=True, explicit_end=True,
default_flow_style=False, allow_unicode=True, version=(1, 2))
except yaml3ed.YAMLError as exc:
print(exc)
def load_yaml(self, template_file):
with open(template_file) as f:
try:
return yaml.safe_load(f)
except yaml.YAMLError as e:
print(e)
return []
def set_env(self, env, config=None):
"""
Loads config from container.yml, and stores the resulting dict to self._config.
:param env: string of either 'dev' or 'prod'. Indicates 'dev_overrides' handling.
:return: None
"""
assert env in ['dev', 'prod']
if not config:
try:
config = yaml.round_trip_load(open(self.config_path))
except IOError:
raise AnsibleContainerNotInitializedException()
except yaml.YAMLError as exc:
raise AnsibleContainerConfigException(u"Parsing container.yml - %s" % text_type(exc))
self._validate_config(config)
for service, service_config in iteritems(config.get('services') or {}):
if not service_config or isinstance(service_config, string_types):
raise AnsibleContainerConfigException(u"Error: no definition found in container.yml for service %s."
% service)
self._update_service_config(env, service_config)
# Insure settings['pwd'] = base_path. Will be used later by conductor to resolve $PWD in volumes.
if config.get('settings', None) is None:
config['settings'] = ordereddict()
config['settings']['pwd'] = self.base_path
self._resolve_defaults(config)
logger.debug(u"Parsed config", config=config)
self._config = config
def _get_variables_from_file(self, var_file):
"""
Looks for file relative to base_path. If not found, checks relative to base_path/ansible.
If file extension is .yml | .yaml, parses as YAML, otherwise parses as JSON.
:return: ruamel.ordereddict
"""
abspath = path.abspath(var_file)
if not path.exists(abspath):
dirname, filename = path.split(abspath)
raise AnsibleContainerConfigException(
u'Variables file "%s" not found. (I looked in "%s" for it.)' % (filename, dirname)
)
logger.debug("Use variable file: %s", abspath, file=abspath)
if path.splitext(abspath)[-1].lower().endswith(('yml', 'yaml')):
try:
config = yaml.round_trip_load(open(abspath))
except yaml.YAMLError as exc:
raise AnsibleContainerConfigException(u"YAML exception: %s" % text_type(exc))
else:
try:
config = json.load(open(abspath))
except Exception as exc:
raise AnsibleContainerConfigException(u"JSON exception: %s" % text_type(exc))
return iteritems(config)
def set_env(self, env, config=None):
try:
config = yaml.round_trip_load(open(self.config_path))
except IOError:
raise AnsibleContainerNotInitializedException()
except yaml.YAMLError as exc:
raise AnsibleContainerConfigException(u"Parsing container.yml - %s" % unicode(exc))
new_services = yaml.compat.ordereddict()
for service_name, service_config in iteritems(config.get('services') or {}):
if service_config.get('containers'):
# If containers is defined, convert it to services, and drop any other keys
for container in service_config['containers']:
if not container.get('container_name'):
raise AnsibleContainerConfigException(
u"Expecting container to have container_name defined. None found."
)
new_service_name = "{}-{}".format(service_name, container['container_name'])
new_services[new_service_name] = copy.deepcopy(container)
else:
new_services[service_name] = copy.deepcopy(service_config)
config['services'] = new_services
super(AnsibleContainerConfig, self).set_env(env, config=config)
if self._config.get('volumes'):
for vol_key in self._config['volumes']:
if 'docker' in self._config['volumes'][vol_key]:
settings = copy.deepcopy(self._config['volumes'][vol_key][self.engine_name])
self._config['volumes'][vol_key] = settings
else:
# remove non-docker settings
for engine_name in self.remove_engines:
if engine_name in self._config['volumes'][vol_key]:
del self._config['volumes'][vol_key][engine_name]
def load(yaml_string, schema=None, label=u"<unicode string>"):
"""
Parse the first YAML document in a string
and produce corresponding YAML object.
"""
if str(type(yaml_string)) not in ("<type 'unicode'>", "<type 'str'>", "<class 'str'>"):
raise TypeError("StrictYAML can only read a string of valid YAML.")
# We manufacture a class that has the label we want
DynamicStrictYAMLLoader = type('DynamicStrictYAMLLoader', (StrictYAMLLoader,), {"label": label})
try:
document = ruamelyaml.load(yaml_string, Loader=DynamicStrictYAMLLoader)
except ruamelyaml.YAMLError as parse_error:
if parse_error.context_mark is not None:
parse_error.context_mark.name = label
if parse_error.problem_mark is not None:
parse_error.problem_mark.name = label
raise parse_error
# Document is just a (string, int, etc.)
if type(document) not in (CommentedMap, CommentedSeq):
document = yaml_string
if schema is None:
schema = Any()
return schema(YAMLChunk(document, label=label))
def parse(filename):
"""
Parse a YAML file
"""
try:
with open(filename) as f:
try:
return yaml.load(f)
except yaml.YAMLError as e:
log.critical("Problem parsing {} as YAML: {}".format(
filename, e))
return None
except FileNotFoundError:
log.critical("Problem opening {}: File was not found".format(filename))
return None
def __init__(self, *args, **kwargs):
if type(self) is BaseProvider:
return
if self.debug:
print(f'{self._typ.upper()} Provider .ctor')
if not self._base_instance:
self._base_instance = self.__class__.__bases__[0]()
base_attributes = inspect.getmembers(self._base_instance, lambda a:not(inspect.isroutine(a)))
base_keys = [a[0] for a in base_attributes if not(a[0].startswith('_'))]
# base_keys = ['debug', 'default_mc_version']
for attribute_key in base_keys:
if attribute_key in kwargs:
value = kwargs.get(attribute_key)
setattr(self, attribute_key, kwargs[attribute_key])
provider_settings = kwargs.get('provider_settings', {})
provider_settings = provider_settings.get(self._typ, {})
if self.debug:
print(f'{self._typ} settings: {provider_settings}')
attributes = inspect.getmembers(self, lambda a:not(inspect.isroutine(a)))
attribute_keys = [a[0] for a in attributes if not(a[0].startswith('_'))]
attribute_keys = list(set(attribute_keys) - set(base_keys))
path = Path(kwargs['data_path'], 'defaults.yaml')
if path.is_dir():
path.rmdir()
global_defaults = {}
if path.exists():
with open(path, 'r') as stream:
try:
global_defaults = yaml.safe_load(stream)
except yaml.YAMLError as exc:
print(exc)
global_defaults = {}
# get all default values
global_defaults[self._typ] = {k: getattr(self, k) for k in attribute_keys}
with open(path, 'w') as outfile:
yaml.dump(global_defaults, outfile, default_flow_style=False)
# write provider settings overrides to self
for attribute_key in attribute_keys:
if attribute_key in provider_settings:
value = provider_settings.get(attribute_key)
if self.debug:
print(f'setting {attribute_key}, value={value}')
setattr(self, attribute_key, provider_settings[attribute_key])
