diff --git a/gcloud/connection.py b/gcloud/connection.py
index f70b1e2720f8..d411b04e4b82 100644
--- a/gcloud/connection.py
+++ b/gcloud/connection.py
@@ -160,7 +160,7 @@ def build_api_url(cls, path, query_params=None,
         return url
 
     def _make_request(self, method, url, data=None, content_type=None,
-                      headers=None):
+                      headers=None, target_object=None):
         """A low level method to send a request to the API.
 
         Typically, you shouldn't need to use this method.
@@ -180,6 +180,12 @@ def _make_request(self, method, url, data=None, content_type=None,
         :type headers: dict
         :param headers: A dictionary of HTTP headers to send with the request.
 
+        :type target_object: object or :class:`NoneType`
+        :param target_object: Argument to be used by library callers.
+                              This can allow custom behavior, for example, to
+                              defer an HTTP request and complete initialization
+                              of the object at a later time.
+
         :rtype: tuple of ``response`` (a dictionary of sorts)
                 and ``content`` (a string).
         :returns: The HTTP response object and the content of the response,
@@ -200,9 +206,10 @@ def _make_request(self, method, url, data=None, content_type=None,
 
         headers['User-Agent'] = self.USER_AGENT
 
-        return self._do_request(method, url, headers, data)
+        return self._do_request(method, url, headers, data, target_object)
 
-    def _do_request(self, method, url, headers, data):
+    def _do_request(self, method, url, headers, data,
+                    target_object):  # pylint: disable=unused-argument
         """Low-level helper:  perform the actual API request over HTTP.
 
         Allows batch context managers to override and defer a request.
@@ -219,6 +226,10 @@ def _do_request(self, method, url, headers, data):
         :type data: string
         :param data: The data to send as the body of the request.
 
+        :type target_object: object or :class:`NoneType`
+        :param target_object: Unused ``target_object`` here but may be used
+                              by a superclass.
+
         :rtype: tuple of ``response`` (a dictionary of sorts)
                 and ``content`` (a string).
         :returns: The HTTP response object and the content of the response.
@@ -229,7 +240,7 @@ def _do_request(self, method, url, headers, data):
     def api_request(self, method, path, query_params=None,
                     data=None, content_type=None,
                     api_base_url=None, api_version=None,
-                    expect_json=True):
+                    expect_json=True, _target_object=None):
         """Make a request over the HTTP transport to the API.
 
         You shouldn't need to use this method, but if you plan to
@@ -274,6 +285,12 @@ def api_request(self, method, path, query_params=None,
                             response as JSON and raise an exception if
                             that cannot be done.  Default is True.
 
+        :type _target_object: object or :class:`NoneType`
+        :param _target_object: Protected argument to be used by library
+                               callers. This can allow custom behavior, for
+                               example, to defer an HTTP request and complete
+                               initialization of the object at a later time.
+
         :raises: Exception if the response code is not 200 OK.
         """
         url = self.build_api_url(path=path, query_params=query_params,
@@ -287,12 +304,14 @@ def api_request(self, method, path, query_params=None,
             content_type = 'application/json'
 
         response, content = self._make_request(
-            method=method, url=url, data=data, content_type=content_type)
+            method=method, url=url, data=data, content_type=content_type,
+            target_object=_target_object)
 
         if not 200 <= response.status < 300:
             raise make_exception(response, content)
 
-        if content and expect_json:
+        string_or_bytes = (six.binary_type, six.text_type)
+        if content and expect_json and isinstance(content, string_or_bytes):
             content_type = response.get('content-type', '')
             if not content_type.startswith('application/json'):
                 raise TypeError('Expected JSON, got %s' % content_type)
diff --git a/gcloud/storage/_helpers.py b/gcloud/storage/_helpers.py
index e6b528b01a12..a5acb45c645e 100644
--- a/gcloud/storage/_helpers.py
+++ b/gcloud/storage/_helpers.py
@@ -60,7 +60,8 @@ def reload(self, connection=None):
         # are handled via custom endpoints.
         query_params = {'projection': 'noAcl'}
         api_response = connection.api_request(
-            method='GET', path=self.path, query_params=query_params)
+            method='GET', path=self.path, query_params=query_params,
+            _target_object=self)
         self._set_properties(api_response)
 
     def _patch_property(self, name, value):
@@ -84,7 +85,7 @@ def _patch_property(self, name, value):
     def _set_properties(self, value):
         """Set the properties for the current object.
 
-        :type value: dict
+        :type value: dict or :class:`gcloud.storage.batch._FutureDict`
         :param value: The properties to be set.
         """
         self._properties = value
@@ -108,7 +109,7 @@ def patch(self, connection=None):
                                  for key in self._changes)
         api_response = connection.api_request(
             method='PATCH', path=self.path, data=update_properties,
-            query_params={'projection': 'full'})
+            query_params={'projection': 'full'}, _target_object=self)
         self._set_properties(api_response)
 
 
diff --git a/gcloud/storage/batch.py b/gcloud/storage/batch.py
index ad811d234184..151e99a18888 100644
--- a/gcloud/storage/batch.py
+++ b/gcloud/storage/batch.py
@@ -20,12 +20,14 @@
 from email.mime.application import MIMEApplication
 from email.mime.multipart import MIMEMultipart
 from email.parser import Parser
+import httplib2
 import io
 import json
 
 import six
 
 from gcloud._helpers import _LocalStack
+from gcloud.exceptions import make_exception
 from gcloud.storage import _implicit_environ
 from gcloud.storage.connection import Connection
 
@@ -71,6 +73,54 @@ class NoContent(object):
     status = 204
 
 
+class _FutureDict(object):
+    """Class to hold a future value for a deferred request.
+
+    Used by for requests that get sent in a :class:`Batch`.
+    """
+
+    @staticmethod
+    def get(key, default=None):
+        """Stand-in for dict.get.
+
+        :type key: object
+        :param key: Hashable dictionary key.
+
+        :type default: object
+        :param default: Fallback value to dict.get.
+
+        :raises: :class:`KeyError` always since the future is intended to fail
+                 as a dictionary.
+        """
+        raise KeyError('Cannot get(%r, default=%r) on a future' % (
+            key, default))
+
+    def __getitem__(self, key):
+        """Stand-in for dict[key].
+
+        :type key: object
+        :param key: Hashable dictionary key.
+
+        :raises: :class:`KeyError` always since the future is intended to fail
+                 as a dictionary.
+        """
+        raise KeyError('Cannot get item %r from a future' % (key,))
+
+    def __setitem__(self, key, value):
+        """Stand-in for dict[key] = value.
+
+        :type key: object
+        :param key: Hashable dictionary key.
+
+        :type value: object
+        :param value: Dictionary value.
+
+        :raises: :class:`KeyError` always since the future is intended to fail
+                 as a dictionary.
+        """
+        raise KeyError('Cannot set %r -> %r on a future' % (key, value))
+
+
 class Batch(Connection):
     """Proxy an underlying connection, batching up change operations.
 
@@ -86,9 +136,9 @@ def __init__(self, connection=None):
         super(Batch, self).__init__()
         self._connection = connection
         self._requests = []
-        self._responses = []
+        self._target_objects = []
 
-    def _do_request(self, method, url, headers, data):
+    def _do_request(self, method, url, headers, data, target_object):
         """Override Connection:  defer actual HTTP request.
 
         Only allow up to ``_MAX_BATCH_SIZE`` requests to be deferred.
@@ -109,22 +159,22 @@ def _do_request(self, method, url, headers, data):
                 and ``content`` (a string).
         :returns: The HTTP response object and the content of the response.
         """
-        if method == 'GET':
-            _req = self._connection.http.request
-            return _req(method=method, uri=url, headers=headers, body=data)
-
         if len(self._requests) >= self._MAX_BATCH_SIZE:
             raise ValueError("Too many deferred requests (max %d)" %
                              self._MAX_BATCH_SIZE)
         self._requests.append((method, url, headers, data))
-        return NoContent(), ''
-
-    def finish(self):
-        """Submit a single `multipart/mixed` request w/ deferred requests.
-
-        :rtype: list of tuples
-        :returns: one ``(status, reason, payload)`` tuple per deferred request.
-        :raises: ValueError if no requests have been deferred.
+        result = _FutureDict()
+        self._target_objects.append(target_object)
+        if target_object is not None:
+            target_object._properties = result
+        return NoContent(), result
+
+    def _prepare_batch_request(self):
+        """Prepares headers and body for a batch request.
+
+        :rtype: tuple (dict, string)
+        :returns: The pair of headers and body of the batch request to be sent.
+        :raises: :class:`ValueError` if no requests have been deferred.
         """
         if len(self._requests) == 0:
             raise ValueError("No deferred requests")
@@ -146,14 +196,51 @@ def finish(self):
 
         # Strip off redundant header text
         _, body = payload.split('\n\n', 1)
-        headers = dict(multi._headers)
+        return dict(multi._headers), body
+
+    def _finish_futures(self, responses):
+        """Apply all the batch responses to the futures created.
+
+        :type responses: list of (headers, payload) tuples.
+        :param responses: List of headers and payloads from each response in
+                          the batch.
+
+        :raises: :class:`ValueError` if no requests have been deferred.
+        """
+        # If a bad status occurs, we track it, but don't raise an exception
+        # until all futures have been populated.
+        exception_args = None
+
+        if len(self._target_objects) != len(responses):
+            raise ValueError('Expected a response for every request.')
+
+        for target_object, sub_response in zip(self._target_objects,
+                                               responses):
+            resp_headers, sub_payload = sub_response
+            if not 200 <= resp_headers.status < 300:
+                exception_args = exception_args or (resp_headers,
+                                                    sub_payload)
+            elif target_object is not None:
+                target_object._properties = sub_payload
+
+        if exception_args is not None:
+            raise make_exception(*exception_args)
+
+    def finish(self):
+        """Submit a single `multipart/mixed` request w/ deferred requests.
+
+        :rtype: list of tuples
+        :returns: one ``(headers, payload)`` tuple per deferred request.
+        """
+        headers, body = self._prepare_batch_request()
 
         url = '%s/batch' % self.API_BASE_URL
 
-        _req = self._connection._make_request
-        response, content = _req('POST', url, data=body, headers=headers)
-        self._responses = list(_unpack_batch_response(response, content))
-        return self._responses
+        response, content = self._connection._make_request(
+            'POST', url, data=body, headers=headers)
+        responses = list(_unpack_batch_response(response, content))
+        self._finish_futures(responses)
+        return responses
 
     @staticmethod
     def current():
@@ -199,7 +286,20 @@ def _generate_faux_mime_message(parser, response, content):
 
 
 def _unpack_batch_response(response, content):
-    """Convert response, content -> [(status, reason, payload)]."""
+    """Convert response, content -> [(headers, payload)].
+
+    Creates a generator of tuples of emulating the responses to
+    :meth:`httplib2.Http.request` (a pair of headers and payload).
+
+    :type response: :class:`httplib2.Response`
+    :param response: HTTP response / headers from a request.
+
+    :type content: string
+    :param content: Response payload with a batch response.
+
+    :rtype: generator
+    :returns: A generator of header, payload pairs.
+    """
     parser = Parser()
     message = _generate_faux_mime_message(parser, response, content)
 
@@ -208,10 +308,13 @@ def _unpack_batch_response(response, content):
 
     for subrequest in message._payload:
         status_line, rest = subrequest._payload.split('\n', 1)
-        _, status, reason = status_line.split(' ', 2)
-        message = parser.parsestr(rest)
-        payload = message._payload
-        ctype = message['Content-Type']
+        _, status, _ = status_line.split(' ', 2)
+        sub_message = parser.parsestr(rest)
+        payload = sub_message._payload
+        ctype = sub_message['Content-Type']
+        msg_headers = dict(sub_message._headers)
+        msg_headers['status'] = status
+        headers = httplib2.Response(msg_headers)
         if ctype and ctype.startswith('application/json'):
             payload = json.loads(payload)
-        yield status, reason, payload
+        yield headers, payload
diff --git a/gcloud/storage/blob.py b/gcloud/storage/blob.py
index ed6a00dda966..dd55525a6e25 100644
--- a/gcloud/storage/blob.py
+++ b/gcloud/storage/blob.py
@@ -227,7 +227,8 @@ def exists(self, connection=None):
             # minimize the returned payload.
             query_params = {'fields': 'name'}
             connection.api_request(method='GET', path=self.path,
-                                   query_params=query_params)
+                                   query_params=query_params,
+                                   _target_object=self)
             return True
         except NotFound:
             return False
diff --git a/gcloud/storage/bucket.py b/gcloud/storage/bucket.py
index efdb7b567c7f..6d1a502d5c95 100644
--- a/gcloud/storage/bucket.py
+++ b/gcloud/storage/bucket.py
@@ -131,8 +131,14 @@ def exists(self, connection=None):
             # We only need the status code (200 or not) so we seek to
             # minimize the returned payload.
             query_params = {'fields': 'name'}
+            # We intentionally pass `_target_object=None` since fields=name
+            # would limit the local properties.
             connection.api_request(method='GET', path=self.path,
-                                   query_params=query_params)
+                                   query_params=query_params,
+                                   _target_object=None)
+            # NOTE: This will not fail immediately in a batch. However, when
+            #       Batch.finish() is called, the resulting `NotFound` will be
+            #       raised.
             return True
         except NotFound:
             return False
@@ -169,7 +175,7 @@ def create(self, project=None, connection=None):
         query_params = {'project': project}
         api_response = connection.api_request(
             method='POST', path='/b', query_params=query_params,
-            data={'name': self.name})
+            data={'name': self.name}, _target_object=self)
         self._set_properties(api_response)
 
     @property
@@ -238,11 +244,13 @@ def get_blob(self, blob_name, connection=None):
         connection = _require_connection(connection)
         blob = Blob(bucket=self, name=blob_name)
         try:
-            response = connection.api_request(method='GET',
-                                              path=blob.path)
-            name = response.get('name')  # Expect this to be blob_name
-            blob = Blob(name, bucket=self)
+            response = connection.api_request(
+                method='GET', path=blob.path, _target_object=blob)
+            # NOTE: We assume response.get('name') matches `blob_name`.
             blob._set_properties(response)
+            # NOTE: This will not fail immediately in a batch. However, when
+            #       Batch.finish() is called, the resulting `NotFound` will be
+            #       raised.
             return blob
         except NotFound:
             return None
@@ -354,7 +362,11 @@ def delete(self, force=False, connection=None):
             self.delete_blobs(blobs, on_error=lambda blob: None,
                               connection=connection)
 
-        connection.api_request(method='DELETE', path=self.path)
+        # We intentionally pass `_target_object=None` since a DELETE
+        # request has no response value (whether in a standard request or
+        # in a batch request).
+        connection.api_request(method='DELETE', path=self.path,
+                               _target_object=None)
 
     def delete_blob(self, blob_name, connection=None):
         """Deletes a blob from the current bucket.
@@ -392,7 +404,11 @@ def delete_blob(self, blob_name, connection=None):
         """
         connection = _require_connection(connection)
         blob_path = Blob.path_helper(self.path, blob_name)
-        connection.api_request(method='DELETE', path=blob_path)
+        # We intentionally pass `_target_object=None` since a DELETE
+        # request has no response value (whether in a standard request or
+        # in a batch request).
+        connection.api_request(method='DELETE', path=blob_path,
+                               _target_object=None)
 
     def delete_blobs(self, blobs, on_error=None, connection=None):
         """Deletes a list of blobs from the current bucket.
@@ -456,7 +472,8 @@ def copy_blob(blob, destination_bucket, new_name=None,
             new_name = blob.name
         new_blob = Blob(bucket=destination_bucket, name=new_name)
         api_path = blob.path + '/copyTo' + new_blob.path
-        copy_result = connection.api_request(method='POST', path=api_path)
+        copy_result = connection.api_request(method='POST', path=api_path,
+                                             _target_object=new_blob)
         new_blob._set_properties(copy_result)
         return new_blob
 
diff --git a/gcloud/storage/test__helpers.py b/gcloud/storage/test__helpers.py
index fabf23d4718b..104ecb38bd05 100644
--- a/gcloud/storage/test__helpers.py
+++ b/gcloud/storage/test__helpers.py
@@ -73,6 +73,13 @@ def test_reload_w_explicit_connection(self):
         # Make sure changes get reset by reload.
         self.assertEqual(derived._changes, set())
 
+    def test__set_properties(self):
+        mixin = self._makeOne()
+        self.assertEqual(mixin._properties, {})
+        VALUE = object()
+        mixin._set_properties(VALUE)
+        self.assertEqual(mixin._properties, VALUE)
+
     def test__patch_property(self):
         derived = self._derivedClass()()
         derived._patch_property('foo', 'Foo')
diff --git a/gcloud/storage/test_batch.py b/gcloud/storage/test_batch.py
index aaa2e94dc10e..94695671090e 100644
--- a/gcloud/storage/test_batch.py
+++ b/gcloud/storage/test_batch.py
@@ -89,7 +89,7 @@ def test_ctor_w_explicit_connection(self):
         batch = self._makeOne(connection)
         self.assertTrue(batch._connection is connection)
         self.assertEqual(len(batch._requests), 0)
-        self.assertEqual(len(batch._responses), 0)
+        self.assertEqual(len(batch._target_objects), 0)
 
     def test_ctor_w_implicit_connection(self):
         from gcloud.storage._testing import _monkey_defaults
@@ -101,92 +101,108 @@ def test_ctor_w_implicit_connection(self):
 
         self.assertTrue(batch._connection is connection)
         self.assertEqual(len(batch._requests), 0)
-        self.assertEqual(len(batch._responses), 0)
+        self.assertEqual(len(batch._target_objects), 0)
 
-    def test__make_request_GET_forwarded_to_connection(self):
+    def test__make_request_GET_normal(self):
+        from gcloud.storage.batch import _FutureDict
         URL = 'http://example.com/api'
         expected = _Response()
         http = _HTTP((expected, ''))
         connection = _Connection(http=http)
         batch = self._makeOne(connection)
-        response, content = batch._make_request('GET', URL)
-        self.assertTrue(response is expected)
-        self.assertEqual(content, '')
+        target = _MockObject()
+        response, content = batch._make_request('GET', URL,
+                                                target_object=target)
+        self.assertEqual(response.status, 204)
+        self.assertTrue(isinstance(content, _FutureDict))
+        self.assertTrue(target._properties is content)
+        self.assertEqual(http._requests, [])
         EXPECTED_HEADERS = [
             ('Accept-Encoding', 'gzip'),
             ('Content-Length', 0),
         ]
-        self.assertEqual(len(http._requests), 1)
-        self.assertEqual(http._requests[0][0], 'GET')
-        self.assertEqual(http._requests[0][1], URL)
-        headers = http._requests[0][2]
+        solo_request, = batch._requests
+        self.assertEqual(solo_request[0], 'GET')
+        self.assertEqual(solo_request[1], URL)
+        headers = solo_request[2]
         for key, value in EXPECTED_HEADERS:
             self.assertEqual(headers[key], value)
-        self.assertEqual(http._requests[0][3], None)
-        self.assertEqual(batch._requests, [])
+        self.assertEqual(solo_request[3], None)
 
     def test__make_request_POST_normal(self):
+        from gcloud.storage.batch import _FutureDict
         URL = 'http://example.com/api'
         http = _HTTP()  # no requests expected
         connection = _Connection(http=http)
         batch = self._makeOne(connection)
-        response, content = batch._make_request('POST', URL, data={'foo': 1})
+        target = _MockObject()
+        response, content = batch._make_request('POST', URL, data={'foo': 1},
+                                                target_object=target)
         self.assertEqual(response.status, 204)
-        self.assertEqual(content, '')
+        self.assertTrue(isinstance(content, _FutureDict))
+        self.assertTrue(target._properties is content)
         self.assertEqual(http._requests, [])
         EXPECTED_HEADERS = [
             ('Accept-Encoding', 'gzip'),
             ('Content-Length', 10),
         ]
-        self.assertEqual(len(batch._requests), 1)
-        self.assertEqual(batch._requests[0][0], 'POST')
-        self.assertEqual(batch._requests[0][1], URL)
-        headers = batch._requests[0][2]
+        solo_request, = batch._requests
+        self.assertEqual(solo_request[0], 'POST')
+        self.assertEqual(solo_request[1], URL)
+        headers = solo_request[2]
         for key, value in EXPECTED_HEADERS:
             self.assertEqual(headers[key], value)
-        self.assertEqual(batch._requests[0][3], {'foo': 1})
+        self.assertEqual(solo_request[3], {'foo': 1})
 
     def test__make_request_PATCH_normal(self):
+        from gcloud.storage.batch import _FutureDict
         URL = 'http://example.com/api'
         http = _HTTP()  # no requests expected
         connection = _Connection(http=http)
         batch = self._makeOne(connection)
-        response, content = batch._make_request('PATCH', URL, data={'foo': 1})
+        target = _MockObject()
+        response, content = batch._make_request('PATCH', URL, data={'foo': 1},
+                                                target_object=target)
         self.assertEqual(response.status, 204)
-        self.assertEqual(content, '')
+        self.assertTrue(isinstance(content, _FutureDict))
+        self.assertTrue(target._properties is content)
         self.assertEqual(http._requests, [])
         EXPECTED_HEADERS = [
             ('Accept-Encoding', 'gzip'),
             ('Content-Length', 10),
         ]
-        self.assertEqual(len(batch._requests), 1)
-        self.assertEqual(batch._requests[0][0], 'PATCH')
-        self.assertEqual(batch._requests[0][1], URL)
-        headers = batch._requests[0][2]
+        solo_request, = batch._requests
+        self.assertEqual(solo_request[0], 'PATCH')
+        self.assertEqual(solo_request[1], URL)
+        headers = solo_request[2]
         for key, value in EXPECTED_HEADERS:
             self.assertEqual(headers[key], value)
-        self.assertEqual(batch._requests[0][3], {'foo': 1})
+        self.assertEqual(solo_request[3], {'foo': 1})
 
     def test__make_request_DELETE_normal(self):
+        from gcloud.storage.batch import _FutureDict
         URL = 'http://example.com/api'
         http = _HTTP()  # no requests expected
         connection = _Connection(http=http)
         batch = self._makeOne(connection)
-        response, content = batch._make_request('DELETE', URL)
+        target = _MockObject()
+        response, content = batch._make_request('DELETE', URL,
+                                                target_object=target)
         self.assertEqual(response.status, 204)
-        self.assertEqual(content, '')
+        self.assertTrue(isinstance(content, _FutureDict))
+        self.assertTrue(target._properties is content)
         self.assertEqual(http._requests, [])
         EXPECTED_HEADERS = [
             ('Accept-Encoding', 'gzip'),
             ('Content-Length', 0),
         ]
-        self.assertEqual(len(batch._requests), 1)
-        self.assertEqual(batch._requests[0][0], 'DELETE')
-        self.assertEqual(batch._requests[0][1], URL)
-        headers = batch._requests[0][2]
+        solo_request, = batch._requests
+        self.assertEqual(solo_request[0], 'DELETE')
+        self.assertEqual(solo_request[1], URL)
+        headers = solo_request[2]
         for key, value in EXPECTED_HEADERS:
             self.assertEqual(headers[key], value)
-        self.assertEqual(batch._requests[0][3], None)
+        self.assertEqual(solo_request[3], None)
 
     def test__make_request_POST_too_many_requests(self):
         URL = 'http://example.com/api'
@@ -223,18 +239,24 @@ def _check_subrequest_payload(self, chunk, method, url, payload):
         lines = chunk.splitlines()
         # blank + 2 headers + blank + request + 2 headers + blank + body
         payload_str = json.dumps(payload)
-        self.assertEqual(len(lines), 9)
         self.assertEqual(lines[0], '')
         self.assertEqual(lines[1], 'Content-Type: application/http')
         self.assertEqual(lines[2], 'MIME-Version: 1.0')
         self.assertEqual(lines[3], '')
         self.assertEqual(lines[4], '%s %s HTTP/1.1' % (method, url))
-        self.assertEqual(lines[5], 'Content-Length: %d' % len(payload_str))
-        self.assertEqual(lines[6], 'Content-Type: application/json')
-        self.assertEqual(lines[7], '')
-        self.assertEqual(json.loads(lines[8]), payload)
+        if method == 'GET':
+            self.assertEqual(len(lines), 7)
+            self.assertEqual(lines[5], '')
+            self.assertEqual(lines[6], '')
+        else:
+            self.assertEqual(len(lines), 9)
+            self.assertEqual(lines[5], 'Content-Length: %d' % len(payload_str))
+            self.assertEqual(lines[6], 'Content-Type: application/json')
+            self.assertEqual(lines[7], '')
+            self.assertEqual(json.loads(lines[8]), payload)
 
     def test_finish_nonempty(self):
+        import httplib2
         URL = 'http://api.example.com/other_api'
         expected = _Response()
         expected['content-type'] = 'multipart/mixed; boundary="DEADBEEF="'
@@ -242,20 +264,24 @@ def test_finish_nonempty(self):
         connection = _Connection(http=http)
         batch = self._makeOne(connection)
         batch.API_BASE_URL = 'http://api.example.com'
-        batch._requests.append(('POST', URL, {}, {'foo': 1, 'bar': 2}))
-        batch._requests.append(('PATCH', URL, {}, {'bar': 3}))
-        batch._requests.append(('DELETE', URL, {}, None))
+        batch._do_request('POST', URL, {}, {'foo': 1, 'bar': 2}, None)
+        batch._do_request('PATCH', URL, {}, {'bar': 3}, None)
+        batch._do_request('DELETE', URL, {}, None, None)
         result = batch.finish()
         self.assertEqual(len(result), len(batch._requests))
-        self.assertEqual(result[0][0], '200')
-        self.assertEqual(result[0][1], 'OK')
-        self.assertEqual(result[0][2], {'foo': 1, 'bar': 2})
-        self.assertEqual(result[1][0], '200')
-        self.assertEqual(result[1][1], 'OK')
-        self.assertEqual(result[1][2], {'foo': 1, 'bar': 3})
-        self.assertEqual(result[2][0], '204')
-        self.assertEqual(result[2][1], 'No Content')
-        self.assertEqual(result[2][2], '')
+        response0 = httplib2.Response({
+            'content-length': '20',
+            'content-type': 'application/json; charset=UTF-8',
+            'status': '200',
+        })
+        self.assertEqual(result[0], (response0, {'foo': 1, 'bar': 2}))
+        response1 = response0
+        self.assertEqual(result[1], (response1, {u'foo': 1, u'bar': 3}))
+        response2 = httplib2.Response({
+            'content-length': '0',
+            'status': '204',
+        })
+        self.assertEqual(result[2], (response2, ''))
         self.assertEqual(len(http._requests), 1)
         method, uri, headers, body = http._requests[0]
         self.assertEqual(method, 'POST')
@@ -279,6 +305,58 @@ def test_finish_nonempty(self):
 
         self._check_subrequest_no_payload(chunks[2], 'DELETE', URL)
 
+    def test_finish_responses_mismatch(self):
+        URL = 'http://api.example.com/other_api'
+        expected = _Response()
+        expected['content-type'] = 'multipart/mixed; boundary="DEADBEEF="'
+        http = _HTTP((expected, _TWO_PART_MIME_RESPONSE_WITH_FAIL))
+        connection = _Connection(http=http)
+        batch = self._makeOne(connection)
+        batch.API_BASE_URL = 'http://api.example.com'
+        batch._requests.append(('GET', URL, {}, None))
+        self.assertRaises(ValueError, batch.finish)
+
+    def test_finish_nonempty_with_status_failure(self):
+        from gcloud.exceptions import NotFound
+        URL = 'http://api.example.com/other_api'
+        expected = _Response()
+        expected['content-type'] = 'multipart/mixed; boundary="DEADBEEF="'
+        http = _HTTP((expected, _TWO_PART_MIME_RESPONSE_WITH_FAIL))
+        connection = _Connection(http=http)
+        batch = self._makeOne(connection)
+        batch.API_BASE_URL = 'http://api.example.com'
+        target1 = _MockObject()
+        target2 = _MockObject()
+        batch._do_request('GET', URL, {}, None, target1)
+        batch._do_request('GET', URL, {}, None, target2)
+        # Make sure futures are not populated.
+        self.assertEqual([future for future in batch._target_objects],
+                         [target1, target2])
+        target2_future_before = target2._properties
+        self.assertRaises(NotFound, batch.finish)
+        self.assertEqual(target1._properties,
+                         {'foo': 1, 'bar': 2})
+        self.assertTrue(target2._properties is target2_future_before)
+
+        self.assertEqual(len(http._requests), 1)
+        method, uri, headers, body = http._requests[0]
+        self.assertEqual(method, 'POST')
+        self.assertEqual(uri, 'http://api.example.com/batch')
+        self.assertEqual(len(headers), 2)
+        ctype, boundary = [x.strip()
+                           for x in headers['Content-Type'].split(';')]
+        self.assertEqual(ctype, 'multipart/mixed')
+        self.assertTrue(boundary.startswith('boundary="=='))
+        self.assertTrue(boundary.endswith('=="'))
+        self.assertEqual(headers['MIME-Version'], '1.0')
+
+        divider = '--' + boundary[len('boundary="'):-1]
+        chunks = body.split(divider)[1:-1]  # discard prolog / epilog
+        self.assertEqual(len(chunks), 2)
+
+        self._check_subrequest_payload(chunks[0], 'GET', URL, {})
+        self._check_subrequest_payload(chunks[1], 'GET', URL, {})
+
     def test_finish_nonempty_non_multipart_response(self):
         URL = 'http://api.example.com/other_api'
         expected = _Response()
@@ -301,29 +379,31 @@ def test_as_context_mgr_wo_error(self):
 
         self.assertEqual(list(_BATCHES), [])
 
+        target1 = _MockObject()
+        target2 = _MockObject()
+        target3 = _MockObject()
         with self._makeOne(connection) as batch:
             self.assertEqual(list(_BATCHES), [batch])
-            batch._make_request('POST', URL, {'foo': 1, 'bar': 2})
-            batch._make_request('PATCH', URL, {'bar': 3})
-            batch._make_request('DELETE', URL)
+            batch._make_request('POST', URL, {'foo': 1, 'bar': 2},
+                                target_object=target1)
+            batch._make_request('PATCH', URL, {'bar': 3},
+                                target_object=target2)
+            batch._make_request('DELETE', URL, target_object=target3)
 
         self.assertEqual(list(_BATCHES), [])
         self.assertEqual(len(batch._requests), 3)
         self.assertEqual(batch._requests[0][0], 'POST')
         self.assertEqual(batch._requests[1][0], 'PATCH')
         self.assertEqual(batch._requests[2][0], 'DELETE')
-        self.assertEqual(len(batch._responses), 3)
-        self.assertEqual(
-            batch._responses[0],
-            ('200', 'OK', {'foo': 1, 'bar': 2}))
-        self.assertEqual(
-            batch._responses[1],
-            ('200', 'OK', {'foo': 1, 'bar': 3}))
-        self.assertEqual(
-            batch._responses[2],
-            ('204', 'No Content', ''))
+        self.assertEqual(batch._target_objects, [target1, target2, target3])
+        self.assertEqual(target1._properties,
+                         {'foo': 1, 'bar': 2})
+        self.assertEqual(target2._properties,
+                         {'foo': 1, 'bar': 3})
+        self.assertEqual(target3._properties, '')
 
     def test_as_context_mgr_w_error(self):
+        from gcloud.storage.batch import _FutureDict
         from gcloud.storage.batch import _BATCHES
         URL = 'http://example.com/api'
         http = _HTTP()
@@ -331,12 +411,17 @@ def test_as_context_mgr_w_error(self):
 
         self.assertEqual(list(_BATCHES), [])
 
+        target1 = _MockObject()
+        target2 = _MockObject()
+        target3 = _MockObject()
         try:
             with self._makeOne(connection) as batch:
                 self.assertEqual(list(_BATCHES), [batch])
-                batch._make_request('POST', URL, {'foo': 1, 'bar': 2})
-                batch._make_request('PATCH', URL, {'bar': 3})
-                batch._make_request('DELETE', URL)
+                batch._make_request('POST', URL, {'foo': 1, 'bar': 2},
+                                    target_object=target1)
+                batch._make_request('PATCH', URL, {'bar': 3},
+                                    target_object=target2)
+                batch._make_request('DELETE', URL, target_object=target3)
                 raise ValueError()
         except ValueError:
             pass
@@ -344,7 +429,12 @@ def test_as_context_mgr_w_error(self):
         self.assertEqual(list(_BATCHES), [])
         self.assertEqual(len(http._requests), 0)
         self.assertEqual(len(batch._requests), 3)
-        self.assertEqual(len(batch._responses), 0)
+        self.assertEqual(batch._target_objects, [target1, target2, target3])
+        # Since the context manager fails, finish will not get called and
+        # the _properties will still be futures.
+        self.assertTrue(isinstance(target1._properties, _FutureDict))
+        self.assertTrue(isinstance(target2._properties, _FutureDict))
+        self.assertTrue(isinstance(target3._properties, _FutureDict))
 
 
 class Test__unpack_batch_response(unittest2.TestCase):
@@ -353,24 +443,58 @@ def _callFUT(self, response, content):
         from gcloud.storage.batch import _unpack_batch_response
         return _unpack_batch_response(response, content)
 
+    def _unpack_helper(self, response, content):
+        import httplib2
+        result = list(self._callFUT(response, content))
+        self.assertEqual(len(result), 3)
+        response0 = httplib2.Response({
+            'content-length': '20',
+            'content-type': 'application/json; charset=UTF-8',
+            'status': '200',
+        })
+        self.assertEqual(result[0], (response0, {u'bar': 2, u'foo': 1}))
+        response1 = response0
+        self.assertEqual(result[1], (response1, {u'foo': 1, u'bar': 3}))
+        response2 = httplib2.Response({
+            'content-length': '0',
+            'status': '204',
+        })
+        self.assertEqual(result[2], (response2, ''))
+
     def test_bytes(self):
         RESPONSE = {'content-type': b'multipart/mixed; boundary="DEADBEEF="'}
         CONTENT = _THREE_PART_MIME_RESPONSE
-        result = list(self._callFUT(RESPONSE, CONTENT))
-        self.assertEqual(len(result), 3)
-        self.assertEqual(result[0], ('200', 'OK', {u'bar': 2, u'foo': 1}))
-        self.assertEqual(result[1], ('200', 'OK', {u'foo': 1, u'bar': 3}))
-        self.assertEqual(result[2], ('204', 'No Content', ''))
+        self._unpack_helper(RESPONSE, CONTENT)
 
     def test_unicode(self):
         RESPONSE = {'content-type': u'multipart/mixed; boundary="DEADBEEF="'}
         CONTENT = _THREE_PART_MIME_RESPONSE.decode('utf-8')
-        result = list(self._callFUT(RESPONSE, CONTENT))
-        self.assertEqual(len(result), 3)
-        self.assertEqual(result[0], ('200', 'OK', {u'bar': 2, u'foo': 1}))
-        self.assertEqual(result[1], ('200', 'OK', {u'foo': 1, u'bar': 3}))
-        self.assertEqual(result[2], ('204', 'No Content', ''))
+        self._unpack_helper(RESPONSE, CONTENT)
+
+
+_TWO_PART_MIME_RESPONSE_WITH_FAIL = b"""\
+--DEADBEEF=
+Content-Type: application/http
+Content-ID: <response-8a09ca85-8d1d-4f45-9eb0-da8e8b07ec83+1>
 
+HTTP/1.1 200 OK
+Content-Type: application/json; charset=UTF-8
+Content-Length: 20
+
+{"foo": 1, "bar": 2}
+
+--DEADBEEF=
+Content-Type: application/http
+Content-ID: <response-8a09ca85-8d1d-4f45-9eb0-da8e8b07ec83+2>
+
+HTTP/1.1 404 Not Found
+Content-Type: application/json; charset=UTF-8
+Content-Length: 35
+
+{"error": {"message": "Not Found"}}
+
+--DEADBEEF=--
+"""
 
 _THREE_PART_MIME_RESPONSE = b"""\
 --DEADBEEF=
@@ -404,6 +528,29 @@ def test_unicode(self):
 """
 
 
+class Test__FutureDict(unittest2.TestCase):
+
+    def _makeOne(self, *args, **kw):
+        from gcloud.storage.batch import _FutureDict
+        return _FutureDict(*args, **kw)
+
+    def test_get(self):
+        future = self._makeOne()
+        self.assertRaises(KeyError, future.get, None)
+
+    def test___getitem__(self):
+        future = self._makeOne()
+        value = orig_value = object()
+        with self.assertRaises(KeyError):
+            value = future[None]
+        self.assertTrue(value is orig_value)
+
+    def test___setitem__(self):
+        future = self._makeOne()
+        with self.assertRaises(KeyError):
+            future[None] = None
+
+
 class _Connection(object):
 
     project = 'TESTING'
@@ -445,3 +592,7 @@ def request(self, method, uri, headers, body):
         self._requests.append((method, uri, headers, body))
         response, self._responses = self._responses[0], self._responses[1:]
         return response
+
+
+class _MockObject(object):
+    pass
diff --git a/gcloud/storage/test_bucket.py b/gcloud/storage/test_bucket.py
index 4f31859c7970..3cbfac8d836f 100644
--- a/gcloud/storage/test_bucket.py
+++ b/gcloud/storage/test_bucket.py
@@ -162,6 +162,7 @@ def api_request(cls, *args, **kwargs):
             'query_params': {
                 'fields': 'name',
             },
+            '_target_object': None,
         }
         expected_cw = [((), expected_called_kwargs)]
         self.assertEqual(_FakeConnection._called_with, expected_cw)
@@ -186,6 +187,7 @@ def api_request(cls, *args, **kwargs):
             'query_params': {
                 'fields': 'name',
             },
+            '_target_object': None,
         }
         expected_cw = [((), expected_called_kwargs)]
         self.assertEqual(_FakeConnection._called_with, expected_cw)
@@ -330,7 +332,11 @@ def test_delete_default_miss(self):
         connection = _Connection()
         bucket = self._makeOne(NAME)
         self.assertRaises(NotFound, bucket.delete, connection=connection)
-        expected_cw = [{'method': 'DELETE', 'path': bucket.path}]
+        expected_cw = [{
+            'method': 'DELETE',
+            'path': bucket.path,
+            '_target_object': None,
+        }]
         self.assertEqual(connection._deleted_buckets, expected_cw)
 
     def test_delete_explicit_hit(self):
@@ -341,7 +347,11 @@ def test_delete_explicit_hit(self):
         bucket = self._makeOne(NAME, connection)
         result = bucket.delete(force=True, connection=connection)
         self.assertTrue(result is None)
-        expected_cw = [{'method': 'DELETE', 'path': bucket.path}]
+        expected_cw = [{
+            'method': 'DELETE',
+            'path': bucket.path,
+            '_target_object': None,
+        }]
         self.assertEqual(connection._deleted_buckets, expected_cw)
 
     def test_delete_explicit_force_delete_blobs(self):
@@ -361,7 +371,11 @@ def test_delete_explicit_force_delete_blobs(self):
         bucket = self._makeOne(NAME, connection)
         result = bucket.delete(force=True, connection=connection)
         self.assertTrue(result is None)
-        expected_cw = [{'method': 'DELETE', 'path': bucket.path}]
+        expected_cw = [{
+            'method': 'DELETE',
+            'path': bucket.path,
+            '_target_object': None,
+        }]
         self.assertEqual(connection._deleted_buckets, expected_cw)
 
     def test_delete_explicit_force_miss_blobs(self):
@@ -374,7 +388,11 @@ def test_delete_explicit_force_miss_blobs(self):
         bucket = self._makeOne(NAME, connection)
         result = bucket.delete(force=True, connection=connection)
         self.assertTrue(result is None)
-        expected_cw = [{'method': 'DELETE', 'path': bucket.path}]
+        expected_cw = [{
+            'method': 'DELETE',
+            'path': bucket.path,
+            '_target_object': None,
+        }]
         self.assertEqual(connection._deleted_buckets, expected_cw)
 
     def test_delete_explicit_too_many(self):