Storage: Blob.exists() does not work within Batch context

storage/google/cloud/storage/blob.py

@@ -56,6 +56,7 @@
 from google.cloud.storage._helpers import _scalar_property
 from google.cloud.storage._signing import generate_signed_url_v2
 from google.cloud.storage._signing import generate_signed_url_v4
+from google.cloud.storage.batch import _FutureDict
 from google.cloud.storage.acl import ACL
 from google.cloud.storage.acl import ObjectACL
 
@@ -466,18 +467,24 @@ def exists(self, client=None):
         query_params["fields"] = "name"
 
         try:
-            # We intentionally pass `_target_object=None` since fields=name
-            # would limit the local properties.
+            if client.current_batch is not None:
+                target = FutureBoolResult()
+                result = target
+            else:
+                # We intentionally pass `_target_object=None` since fields=name
+                # would limit the local properties.
+                target, result = None, True
+
             client._connection.api_request(
                 method="GET",
                 path=self.path,
                 query_params=query_params,
-                _target_object=None,
+                _target_object=target,
             )
             # NOTE: This will not fail immediately in a batch. However, when
             #       Batch.finish() is called, the resulting `NotFound` will be
             #       raised.
-            return True
+            return result
         except NotFound:
             return False
 
@@ -2057,3 +2064,29 @@ def _add_query_parameters(base_url, name_value_pairs):
     query = parse_qsl(query)
     query.extend(name_value_pairs)
     return urlunsplit((scheme, netloc, path, urlencode(query), frag))
+
+
+class FutureBoolResult(object):
+    """Represents the result of future api request.
+
+    This can be used as a target object for the future request
+    (which is part of batch mechanism) to get result of this request.
+    """
+    _properties = {}
+
+    @property
+    def _is_successfull(self):
+        """Checks if binded future request have been completed without any exception."""
+        return not isinstance(self._properties, _FutureDict)
+
+    def __bool__(self):
+        return self._is_successfull
+
+    def __repr__(self):
+        return str(self._is_successfull)
+
+    def __eq__(self, other):
+        return self._is_successfull == other
+
+    def __ne__(self, other):
+        return self._is_successfull != other

storage/tests/unit/test_blob.py

@@ -20,13 +20,45 @@
 import os
 import tempfile
 import unittest
+import requests
 
 import google.cloud.storage.blob
 import mock
 import six
 from six.moves import http_client
 
 
+def _make_response(status_code=http_client.OK, content=b"", headers={}):
+    response = requests.Response()
+    response.status_code = status_code
+    response._content = content
+    response.headers = headers
+    response.request = requests.Request()
+    return response
+
+
+def _make_connection():
+    from google.cloud.storage._http import Connection
+
+    mock_conn = mock.create_autospec(Connection)
+    mock_conn.user_agent = "testing 1.2.3"
+    mock_conn._make_request.return_value = _make_response(
+        status_code=200,
+        headers={
+            "content-type": "multipart/mixed; boundary=batch_pK7JBAk73-E=_AA5eFwv4m2Q=",
+        },
+        content=b"""
+--batch_pK7JBAk73-E=_AA5eFwv4m2Q=
+HTTP/1.1 404 NotFound
+
+--batch_pK7JBAk73-E=_AA5eFwv4m2Q=
+HTTP/1.1 200 OK
+
+--batch_pK7JBAk73-E=_AA5eFwv4m2Q=--"""
+    )
+    return mock_conn
+
+
 def _make_credentials():
     import google.auth.credentials
 
@@ -582,6 +614,27 @@ def test_exists_miss(self):
             },
         )
 
+    def test_exists_miss_within_batch(self):
+        from google.cloud.storage.client import Client
+        from google.cloud.exceptions import NotFound
+
+        client = Client(credentials=_make_credentials())
+        client._base_connection = _make_connection()
+
+        bucket = _Bucket(client)
+        blob1 = self._make_one("nonesuch1", bucket=bucket)
+        blob2 = self._make_one("nonesuch2", bucket=bucket)
+
+        try:
+            with client.batch():
+                bool1 = blob1.exists()
+                bool2 = blob2.exists()
+        except NotFound:
+            pass
+
+        self.assertFalse(bool1)
+        self.assertTrue(bool2)
+
     def test_exists_hit_w_user_project(self):
         BLOB_NAME = "blob-name"
         USER_PROJECT = "user-project-123"
@@ -3311,6 +3364,7 @@ def delete_blob(self, blob_name, client=None, generation=None):
 class _Client(object):
     def __init__(self, connection):
         self._base_connection = connection
+        self.current_batch = None
 
     @property
     def _connection(self):