Parse volumes key.

        Args:
            volume_values (dict): volume configuration values

        Returns:
            string, volume specification with mount source and container path
192154
def _parse_volumes(volume_values: dict) -> str:
        
        for v_values in volume_values:
            for v_key, v_value in v_values.items():
                if v_key == 'source':
                    if v_value == '.':
                        source = os.path.dirname(
                            os.path.abspath(__file__))
                    else:
                        source = v_value
                if v_key == 'target':
                    target = v_value
            volume_spec = [source + ':' + target]
            return volume_spec
604364
def describe_volumes(self, *volume_ids):
        """"""
        volumeset = {}
        for pos, volume_id in enumerate(volume_ids):
            volumeset["VolumeId.%d" % (pos + 1)] = volume_id
        query = self.query_factory(
            action="DescribeVolumes", creds=self.creds, endpoint=self.endpoint,
            other_params=volumeset)
        d = query.submit()
        return d.addCallback(self.parser.describe_volumes)
1025180
def get_volumes(instance):
    """"""
    if instance.cloud == 'aws':
        client = boto3.client('ec2', instance.region)
        devices = client.describe_instance_attribute(
            InstanceId=instance.id, Attribute='blockDeviceMapping').get('BlockDeviceMappings', [])
        volumes = client.describe_volumes(VolumeIds=[device['Ebs']['VolumeId']
                                                     for device in devices if device.get('Ebs', {}).get('VolumeId')]).get('Volumes', [])
        return {volume['Attachments'][0]['Device']: {'size': volume['Size'], 'volume_type': volume['VolumeType']} for volume in volumes}
    if instance.cloud == 'gcp':
        credentials = GoogleCredentials.get_application_default()
        compute = discovery.build('compute', 'v1', credentials=credentials)
        volumes = {}
        for disk in compute.instances().get(instance=instance.id,
                                            zone=instance.zone,
                                            project=instance.project).execute()['disks']:
            index = disk['index']
            name = disk['deviceName'] if disk['deviceName'] not in [u'persistent-disk-0', 'boot'] else instance.id
            if 'local-ssd' in disk['deviceName']:
                size = 375.0
            if 'local-ssd' in disk['deviceName']:
                size = 375.0
                disk_type = 'local-ssd'
            else:
                disk_data = compute.disks().get(disk=name,
                                                zone=instance.zone,
                                                project=instance.project).execute()
                size = float(disk_data['sizeGb'])
                disk_type = 'pd-ssd'
            volumes[index] = {'size': size,
                              'type': disk['type'],
                              'deviceName': disk['deviceName'],
                              'interface': disk['interface'],
                              'diskType': disk_type}
        return volumes
    raise ValueError('Unknown cloud %s' % instance.cloud)
267911
def _volume_get(self, volume_id):
        '''
        
        '''
        if self.volume_conn is None:
            raise SaltCloudSystemExit('No cinder endpoint available')
        nt_ks = self.volume_conn
        volume = nt_ks.volumes.get(volume_id)
        response = {'name': volume.display_name,
                    'size': volume.size,
                    'id': volume.id,
                    'description': volume.display_description,
                    'attachments': volume.attachments,
                    'status': volume.status
                    }
        return response
188433
def volumes(self):
        
        return sys_volumes.VolumeCollection(
            self._conn, utils.get_subresource_path_by(self, 'Volumes'),
            redfish_version=self.redfish_version)
1090712
async def SetVolumeAttachmentInfo(self, volume_attachments):
        '''
        
        '''
        # map input types to rpc msg
        _params = dict()
        msg = dict(type='StorageProvisioner',
                   request='SetVolumeAttachmentInfo',
                   version=4,
                   params=_params)
        _params['volume-attachments'] = volume_attachments
        reply = await self.rpc(msg)
        return reply
182372
def extract_formats(config_handle):
    
    configurations = dict(config_handle)
    formats = dict(configurations.get('formats', {}))
    return formats
1269933
def _convert_volume_from(self, volume_from):
        
        if ':' in volume_from:
            container, permissions = volume_from.split(':')
        else:
            container = volume_from
            permissions = 'rw'

        if permissions not in ('ro', 'rw'):
            raise ValueError("only permissions supported for volumes_from are rw and ro.")

        return "{0}:{1}".format(container, permissions)
382304
def get_volumes_for_container(self, container_id, skip_empty_source=True):
        
        logger.info("listing volumes for container '%s'", container_id)
        inspect_output = self.d.inspect_container(container_id)
        volumes = inspect_output['Mounts'] or []
        volume_names = [x['Name'] for x in volumes]

        if skip_empty_source:
            # Don't show volumes which are not on the filesystem
            volume_names = [x['Name'] for x in volumes if x['Source'] != ""]

        logger.debug("volumes = %s", volume_names)
        return volume_names
308943
def inspect(self):
        
        if not self.client_config.features['volumes']:
            raise ValueError("Client does not support volume configuration.", self.client_name)
        config_id = self.config_id
        parent_name = config_id.config_name if self.container_map.use_attached_parent_name else None
        volume_name = self.volume_name = self.policy.aname(config_id.map_name, config_id.instance_name,
                                                           parent_name=parent_name)
        if volume_name in self.policy.volume_names[self.client_name]:
            self.detail = self.client.inspect_volume(volume_name)
        else:
            self.detail = NOT_FOUND
827161
def get_volume_object_info(self, location):
        
        param = {'location': location}
        data = self._api.get(url=self._URL['object'].format(
            id=self.id), params=param).json()
        return VolumeObject(api=self._api, **data)
709089
def Parse(self, cmd, args, stdout, stderr, return_val, time_taken,
            knowledge_base):
    """"""
    _ = stderr, time_taken, args, knowledge_base  # Unused.
    self.CheckReturn(cmd, return_val)
    result = rdf_protodict.AttributedDict()
    for entry in self._field_parser.ParseEntries(stdout):
      line_str = " ".join(entry)
      mount_rslt = self.mount_re.match(line_str)
      if mount_rslt:
        device, mount_point, fs_type, option_str = mount_rslt.groups()
        result = rdf_client_fs.Filesystem()
        result.device = device
        result.mount_point = mount_point
        result.type = fs_type
        # Parse these options as a dict as some items may be key/values.
        # KeyValue parser uses OrderedDict as the native parser method. Use it.
        options = KeyValueParser(term=",").ParseToOrderedDict(option_str)
        # Keys without values get assigned [] by default. Because these keys are
        # actually true, if declared, change any [] values to True.
        for k, v in iteritems(options):
          options[k] = v or [True]
        result.options = rdf_protodict.AttributedDict(**options)
        yield result
775315
def parse(self, data):
        # type: (bytes) -> None
        '''
        
        '''
        if self._initialized:
            raise pycdlibexception.PyCdlibInternalError('UDF Logical Volume Implementation Use already initialized')

        (impl_id, self.num_files, self.num_dirs, self.min_udf_read_revision,
         self.min_udf_write_revision,
         self.max_udf_write_revision) = struct.unpack_from(self.FMT, data, 0)

        self.impl_id = UDFEntityID()
        self.impl_id.parse(impl_id)

        self.impl_use = data[46:]

        self._initialized = True
1152938
def get_volume_details(self, volume_name: str) -> dict:
        
        if volume_name not in self.volumes:
            raise RuntimeError('No such volume found: ', volume_name)

        volume = self._client.volumes.get(volume_name)
        return volume.attrs
558782
def volumes(self):
        
        self.connection._check_login()
        response = self.connection._do_get("{}/{}".format(self.connection._api_url, "types/Volume/instances")).json()
        all_volumes = []
        for volume in response:
            all_volumes.append(
                SIO_Volume.from_dict(volume)
            )
        return all_volumes
709497
def parse_client_cert_pair(config_value):
    
    if not config_value:
        return
    client_cert = config_value.split(':')
    if len(client_cert) != 2:
        tips = ('client_cert should be formatted like '
                '"/path/to/cert.pem:/path/to/key.pem"')
        raise ValueError('{0!r} is invalid.\n{1}'.format(config_value, tips))
    return tuple(client_cert)
152539
def get_keys(self, key=None):
        
        if key:
            value = _lxc.Container.get_keys(self, key)
        else:
            value = _lxc.Container.get_keys(self)

        if value is False:
            return False
        elif value.endswith("\n"):
            return value.rstrip("\n").split("\n")
        else:
            return value