def import_config(args, input_file=None):
if not input_file:
input_file = sys.stdin
source = input_file.read().strip()
if source[0] == '{':
# JSON input
config = json.loads(source)
else:
# YAML input
config = yaml.round_trip_load(source)
STATE['stages'] = config['stages']
config['config'] = _encrypt_dict(config['config'])
with open(args.config, 'wt') as f:
if config:
yaml.round_trip_dump(config, f)
python类round_trip_load()的实例源码
def export_config(args, output_file=None):
if not output_file:
output_file = sys.stdout
if os.path.exists(args.config):
with open(args.config, 'rt') as f:
config = yaml.round_trip_load(f.read())
STATE['stages'] = config['stages']
config['config'] = _decrypt_dict(config['config'])
else:
config = {
'stages': {
env['name']: {
'environment': env['name'],
'key': 'enter-key-name-here'
} for env in STATE['awscreds'].environments
},
'config': {}}
if args.json:
output_file.write(json.dumps(config, indent=4))
elif config:
yaml.round_trip_dump(config, output_file)
def _process_section(self, section_value, callback=None, templar=None):
if not templar:
templar = self._templar
processed = ordereddict()
for key, value in section_value.items():
if isinstance(value, string_types):
# strings can be templated
processed[key] = templar.template(value)
if isinstance(processed[key], AnsibleUnsafeText):
processed[key] = str(processed[key])
elif isinstance(value, (list, dict)):
# if it's a dimensional structure, it's cheaper just to serialize
# it, treat it like a template, and then deserialize it again
buffer = BytesIO() # use bytes explicitly, not unicode
yaml.round_trip_dump(value, buffer)
processed[key] = yaml.round_trip_load(
templar.template(buffer.getvalue())
)
else:
# ints, booleans, etc.
processed[key] = value
if callback:
callback(processed)
return processed
def load_yaml(cls, filename: str):
"""
Load a hierarchy of sparse objects from a YAML file.
:param filename: The filename to open.
:return: The configuration object hierarchy
"""
def unpack(mapping, parent=None):
"""
Recursively create Configuration objects with the parent
correctly set, returning the top-most parent.
"""
if mapping is None:
return None
children = config = None
if not isinstance(mapping, cls):
children = mapping.pop('children', None)
config = cls(**cls._coerce_types(mapping), parent=parent)
if children is not None and len(children) > 0:
for child in children:
unpack(child, parent=config)
return config
if filename in cls._yaml_cache:
return cls._yaml_cache[filename]
data = None
with open(filename, 'r') as yaml_file:
data = unpack(yaml.round_trip_load(yaml_file.read()))
if data is not None:
cls._yaml_cache[filename] = data
return data
def get_content_from_role(role_name, relative_path):
role_path = resolve_role_to_path(role_name)
metadata_file = os.path.join(role_path, relative_path)
if os.path.exists(metadata_file):
with open(metadata_file) as ifs:
metadata = yaml.round_trip_load(ifs)
return metadata or yaml.compat.ordereddict()
return yaml.compat.ordereddict()
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_config(filename=DOT_CONFIG_YML, roundtrip=False, fixup=True):
cfg = {}
if os.path.isfile(filename):
try:
with open(filename, 'r') as f:
if roundtrip:
cfg = yaml.round_trip_load(f.read())
else:
cfg = yaml.safe_load(f.read())
if fixup:
cfg = _fixup(cfg)
except Exception as ex:
print('ex =', ex)
raise ConfigLoadError(filename, errors=[ex])
return AttrDict(cfg)
