STORAGE: Bucket.delete_blobs() should use Batch

This will also allow Bucket.delete(force=True) to use Batch()

Ugh! While conceptually this is a relatively simple change, it is a really messy thing to try to test (batching means kicking down a layer below api_request, which makes all the affected tests barf). I've discarded the branch I started.

Good to know. How do you mean "kicking down"?

Batch.finish calls _make_request on its connection, rather than api_request. The existing stub connection doesn't have that, and its contract is different enough that I backed off of trying to add it. It was also going to be tricky to write assertions about the multipart sub-requests (I started factoring out helper methods from TestBatch, but dropped that as well).

No reason it couldn't be done: I just felt like I was down the rabbit hole, and bailed.

Cool. Seems like we could improve the code based on those observations. I'll ping you if I pick this up at some point.

Should this have label: bug?

The following code:

def delete_tmp_files(credentials, tmp_dir):
    logging.info('Deleting temporary files')
    parsed = urlparse.urlparse(tmp_dir)
    client = storage.Client(credentials=credentials)
    bucket = client.bucket(parsed.netloc)
    blobs = bucket.list_blobs(prefix=parsed.path)
    with client.batch():
        bucket.delete_blobs(blobs)

raises error KeyError: "Cannot get('nextPageToken', default=None) on a future" because the iterator returned by list_blobs is making calls to get the next page in the context of a batch.

Maybe this is a separate errors? Seems like the above code should work without encasing blobs in list(blobs)

It's certainly separate from this issue, but good to know. I recommend against passing an iterator like that to delete_blobs() but it is a bug we can try to take a stab at.

Slightly connected to this bug, since that pattern would be unnecessary if delete_blobs used a batch. But I agree it is ~~seperate~~ separate.

Iterating over a collection while mutating it is always problematic, even without adding RPC calls to the mix: see PEP 3106.

I would say the list(blobs) fix is the Right Thing(TM).

In my mind delete_blobs should take an iterator. A Blob object's state is not intrinsically tied to the state of a blob in the cloud, therefor deleting a blob isn't (or shouldn't be) a mutation of the state of the blob itself. The blob is just a wrapper for information about how to make the RPC call.

In my mind delete_blobs should take an iterator.

It does. We're just recommending to you that mutating a Python iterator while using it should be avoided when possible.

Again, how is this a mutation? delete_blob does do any modification of the Blob object. You can't infer "No mutation of any state at all" from that PEP, it's mutation of state associated with objects in the iterator?

Came here while trying to open a new issue concerning this. My usecase is deleting or moving millions of blobs. Currently I use one main thread that lists all blobs in a bucket and puts the ones to be deleted on a queue. A few dozen worker threads then individually delete each blob. I'm averaging about 60 deletes/sec, adding more threads doesn't help. And that's a bit slow for millions of blobs (takes days to run).

I'm trying to use the Batch class for this, but it is not entirely clear to me how to correctly use it and if it even supports deletes, am getting a header parsing error when I try to batch deletes.

It would be great if the bucket.delete_blobs() method (and other related methods for that matter) would use a batch by default. The current code is:

    def delete_blobs(self, blobs, on_error=None, client=None):
        """Deletes a list of blobs from the current bucket.

        Uses :meth:`delete_blob` to delete each individual blob.

        If :attr:`user_project` is set, bills the API request to that project.

        :type blobs: list
        :param blobs: A list of :class:`~google.cloud.storage.blob.Blob`-s or
                      blob names to delete.

        :type on_error: callable
        :param on_error: (Optional) Takes single argument: ``blob``. Called
                         called once for each blob raising
                         :class:`~google.cloud.exceptions.NotFound`;
                         otherwise, the exception is propagated.

        :type client: :class:`~google.cloud.storage.client.Client`
        :param client: (Optional) The client to use.  If not passed, falls back
                       to the ``client`` stored on the current bucket.

        :raises: :class:`~google.cloud.exceptions.NotFound` (if
                 `on_error` is not passed).
        """
        for blob in blobs:
            try:
                blob_name = blob
                if not isinstance(blob_name, six.string_types):
                    blob_name = blob.name
                self.delete_blob(blob_name, client=client)
            except NotFound:
                if on_error is not None:
                    on_error(blob)
                else:
                    raise

The batch version might look something like (just a draft, not tested):

def delete_blobs(self, blobs, on_error=None, client=None):
    with self.client.batch() as batch:
        for blob in blobs:
            try:
                blob_name = blob
                if not isinstance(blob_name, six.string_types):
                    blob_name = blob.name
                self.delete_blob(blob_name, client=client)
            # Send batch off if full
            if len(batch._requests) > batch._MAX_BATCH_SIZE - 1:
                batch.finish()