  def _ListByLabel(
      self,
      labels_filter: Dict[str, str],
      service_object: 'googleapiclient.discovery.Resource',
      filter_union: bool) -> Dict[str, Any]:
    """List Disks/VMs in a project with one/all of the provided labels.

    Private method used to select different compute resources by labels.

    Args:
      labels_filter (Dict[str, str]): A Dict of labels to find e.g.
          {'id': '123'}.
      service_object (googleapiclient.discovery.Resource): Google Compute Engine
          (Disk | Instance) service object.
      filter_union (bool): A boolean; True to get the union of all filters,
          False to get the intersection.

    Returns:
      Dict[str, GoogleComputeInstance|GoogleComputeDisk]: Dictionary mapping
          instances/disks to their respective GoogleComputeInstance /
          GoogleComputeDisk object.

    Raises:
      TypeError: If filter_union is not of type bool
      RuntimeError: If the operation doesn't complete on GCP.
    """

    if not isinstance(filter_union, bool):
      raise TypeError(
          'Filter_union parameter must be of Type boolean. {0:s} '
          'is an invalid argument.'.format(filter_union))

    # pylint: disable=line-too-long
    resource_dict = {
    }  # type: Dict[str, Union[GoogleComputeInstance, GoogleComputeDisk]]
    # pylint: enable=line-too-long
    filter_expression = ''
    operation = 'AND' if filter_union else 'OR'
    for key, value in labels_filter.items():
      filter_expression += 'labels.{0:s}={1:s} {2:s} '.format(
          key, value, operation)
    filter_expression = filter_expression[:-(len(operation) + 1)]

    request = service_object.aggregatedList(
        project=self.project_id, filter=filter_expression)
    while request is not None:
      response = request.execute()

      for item in response['items'].items():
        region_or_zone_string, resource_scoped_list = item

        if 'warning' not in resource_scoped_list.keys():
          _, zone = region_or_zone_string.rsplit('/', 1)
          # Only one of the following loops will execute since the method is
          # called either with a service object Instances or Disks.
          for resource in resource_scoped_list.get('instances', []):
            name = resource['name']
            resource_dict[name] = GoogleComputeInstance(
                self.project_id, zone, name, labels=resource['labels'])

          for resource in resource_scoped_list.get('disks', []):
            name = resource['name']
            resource_dict[name] = GoogleComputeDisk(
                self.project_id, zone, name, labels=resource['labels'])

      request = service_object.aggregatedList_next(
          previous_request=request, previous_response=response)
    return resource_dict
List Disks/VMs in a project with one/all of the provided labels.

Private method used to select different compute resources by labels.

Args:
labels_filter (Dict[str, str]): A Dict of labels to find e.g.
{'id': '123'}.
service_object (googleapiclient.discovery.Resource): Google Compute Engine
(Disk | Instance) service object.
filter_union (bool): A boolean; True to get the union of all filters,
False to get the intersection.

Returns:
Dict[str, GoogleComputeInstance|GoogleComputeDisk]: Dictionary mapping
instances/disks to their respective GoogleComputeInstance /
GoogleComputeDisk object.

Raises:
TypeError: If filter_union is not of type bool
RuntimeError: If the operation doesn't complete on GCP.
libcloudforensics/providers/gcp/internal/compute.py