def save_yaml(self, filename: str=None):
"""
Serialize the hierarchy to a file.
:param filename: Target filename, autogenerated if None
"""
if filename is None:
filename = self.config_filename
with tempfile.NamedTemporaryFile('w', dir=os.path.dirname(filename),
delete=False) as temp:
header = self._yaml_header()
if header is not None:
temp.write(header)
yaml.round_trip_dump(self, stream=temp)
tempname = temp.name
os.rename(tempname, filename)
if filename in self.__class__._yaml_cache:
del self.__class__._yaml_cache[filename]
# YAML library configuration
python类round_trip_dump()的实例源码
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)
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 yaml(self) -> str:
"""
Get the YAML representation of this object as a string
"""
return yaml.round_trip_dump(self)
def _write_config(filename, cfg, roundtrip=False):
try:
with open(filename, 'w') as f:
if roundtrip:
f.write(yaml.round_trip_dump(dict(cfg), indent=4))
else:
f.write(yaml.dump(cfg, indent=4))
except Exception as ex:
raise ConfigWriteError(filename, errors=[ex])
return cfg
def create_proj(self):
yml = CommentedMap()
yml['project'] = CommentedMap()
#
def _add(name):
items = getattr(self, name)
#if BuildItem.trivial_item(items):
# yml['project'][name] = "_default_"
#elif BuildItem.no_flags_in_collection(items):
if BuildItem.no_flags_in_collection(items):
out = []
for s in items:
out.append(s.name)
yml['project'][name] = out
else:
out = []
for s in items:
cm = CommentedMap()
cm[s.name] = CommentedMap()
s.save_config(cm[s.name])
out.append(cm)
yml['project'][name] = out
#
_add('systems')
_add('architectures')
_add('compilers')
_add('build_types')
_add('variants')
txt = yaml.round_trip_dump(yml)
fn = self.kwargs['output_file']
if not os.path.isabs(fn):
fn = os.path.join(self.root_dir, fn)
with open(fn, "w") as f:
f.write(txt)
def write_template(self, output, filename=None):
if not filename:
filename = 'swagger.yml'
swagger_file = os.path.join(self.project_dir, filename)
_, ext = os.path.splitext(filename)
with open(swagger_file, 'w') as fh:
# Could be `.yaml` or `.yml` :/
if '.y' in ext:
fh.write(yaml.round_trip_dump(output))
elif '.json' in ext:
fh.write(json.dumps(output))
return swagger_file
def decrypt(config, output=False):
stage = config['stage']
check_encryption_required_fields(config['stages'][stage])
region = config['stages'][stage]['keyRegion']
kms = boto3.client('kms', region_name=region)
enc_config = get_secret_config(config, stage)
if isinstance(enc_config, string_types):
# This is the old-style secretConfig, when everything was encrypted
# into a single string.
stage_cfg = base64.b64decode(enc_config)
resp = kms.decrypt(CiphertextBlob=stage_cfg)
plain = json.loads(resp['Plaintext'])
if output:
print('Decrypted config for stage {}:\n\n{}'.format(
stage,
yaml.round_trip_dump(plain)))
return plain
elif isinstance(enc_config, dict):
# This is the new way, where all config items are encrypted separately.
plain = {}
for key, value in enc_config.items():
if is_value_already_encrypted(value):
ciphertext_blob = base64.b64decode(
value[len(ENCRYPTED_PREFIX):])
resp = kms.decrypt(CiphertextBlob=ciphertext_blob)
plain[key] = resp['Plaintext']
else:
raise Exception('Found unencrypted item in secretConfig: '
'{}'.format(key))
if output:
print('Decrypted config for stage {}:\n\n{}'.format(
stage,
yaml.round_trip_dump(plain)))
return plain
def render_config(args, output_file=None):
if not output_file:
output_file = sys.stdout
stage = args.stage
env = args.environment or stage
with open(args.config, 'rt') as f:
config = yaml.safe_load(f.read())
STATE['stages'] = config['stages']
config['config'] = _decrypt_item(config['config'], stage=stage, key='',
render=True)
if args.json or args.encrypt or args.python:
rendered_config = json.dumps(
config['config'], indent=None if args.encrypt else 4,
separators=(',', ':') if args.encrypt else (',', ': '))
else:
buf = StringIO()
yaml.round_trip_dump(config['config'], buf)
rendered_config = buf.getvalue()
if args.encrypt or args.python:
STATE['stages'] = config['stages']
encrypted_config = []
while rendered_config:
buffer = _encrypt_text(rendered_config[:4096], env)
rendered_config = rendered_config[4096:]
encrypted_config.append(buffer)
if not args.python:
rendered_config = json.dumps(encrypted_config)
else:
rendered_config = '''ENCRYPTED_CONFIG = {}
import base64
import boto3
import json
def load_config():
config_json = ''
kms = boto3.client('kms')
for buffer in ENCRYPTED_CONFIG:
r = kms.decrypt(CiphertextBlob=base64.b64decode(buffer.encode(
'utf-8')))
config_json += r['Plaintext'].decode('utf-8')
return json.loads(config_json)
CONFIG = load_config()
'''.format(encrypted_config)
output_file.write(rendered_config)
def json2cwl(GATK_json, cwl_dir, cmd_line_options):
"""
Make a cwl file with a given GATK json file in the cwl directory
"""
skeleton_cwl = {
'id': GATK_json['name'],
'cwlVersion': 'v1.0',
'baseCommand': ['java', '-jar', cmd_line_options.gatk_location],
'class': 'CommandLineTool',
'requirements': [
{
"class": "ShellCommandRequirement"
},
{
"class": "InlineJavascriptRequirement",
"expressionLib": [
# Allows you to add annotations
"""function parseTags(param, tags){
if(tags == undefined){
return ' ' + param
}
else{
return ':' + tags.join(',') + ' ' + param
}
}""".replace(" ", "").replace("\n", "")
]
},
{
"class": "DockerRequirement",
"dockerPull": cmd_line_options.docker_container_name
}
]
}
# Create and write the cwl file
fname = GATK_json['name'] + '.cwl'
f = open(os.path.join(cwl_dir, fname), 'a')
cwl_generator(
GATK_json,
skeleton_cwl,
cmd_line_options
)
yaml.round_trip_dump(skeleton_cwl, f) # write the file
f.close()
