Python ruamel.yaml 模块，round_trip_dump() 实例源码

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

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()