diff --git a/core/google/cloud/iterator.py b/core/google/cloud/iterator.py
index b7652e647767..3721f257151f 100644
--- a/core/google/cloud/iterator.py
+++ b/core/google/cloud/iterator.py
@@ -17,17 +17,21 @@
 These iterators simplify the process of paging through API responses
 where the response is a list of results with a ``nextPageToken``.
 
-To make an iterator work, just override the ``get_items_from_response``
-method so that given a response (containing a page of results) it parses
-those results into an iterable of the actual objects you want::
+To make an iterator work, just override the ``PAGE_CLASS`` class
+attribute so that given a response (containing a page of results) can
+be parsed into an iterable page of the actual objects you want::
+
+  class MyPage(Page):
+
+      def _item_to_value(self, item):
+          my_item = MyItemClass(other_arg=True)
+          my_item._set_properties(item)
+          return my_item
+
 
   class MyIterator(Iterator):
-      def get_items_from_response(self, response):
-          items = response.get('items', [])
-          for item in items:
-              my_item = MyItemClass(other_arg=True)
-              my_item._set_properties(item)
-              yield my_item
+
+      PAGE_CLASS = MyPage
 
 You then can use this to get **all** the results from a resource::
 
@@ -38,25 +42,114 @@ def get_items_from_response(self, response):
 you find what you're looking for (resulting in possibly fewer
 requests)::
 
-    >>> for item in MyIterator(...):
-    >>>     print(item.name)
-    >>>     if not item.is_valid:
-    >>>         break
+    >>> for my_item in MyIterator(...):
+    ...     print(my_item.name)
+    ...     if not my_item.is_valid:
+    ...         break
+
+When iterating, not every new item will send a request to the server.
+To monitor these requests, track the current page of the iterator::
+
+    >>> iterator = MyIterator(...)
+    >>> iterator.page_number
+    0
+    >>> next(iterator)
+    <MyItemClass at 0x7f1d3cccf690>
+    >>> iterator.page_number
+    1
+    >>> iterator.page.remaining
+    1
+    >>> next(iterator)
+    <MyItemClass at 0x7f1d3cccfe90>
+    >>> iterator.page.remaining
+    0
+    >>> next(iterator)
+    <MyItemClass at 0x7f1d3cccffd0>
+    >>> iterator.page_number
+    2
+    >>> iterator.page.remaining
+    19
 """
 
 
 import six
 
 
+class Page(object):
+    """Single page of results in an iterator.
+
+    :type parent: :class:`Iterator`
+    :param parent: The iterator that owns the current page.
+
+    :type response: dict
+    :param response: The JSON API response for a page.
+    """
+
+    ITEMS_KEY = 'items'
+
+    def __init__(self, parent, response):
+        self._parent = parent
+        items = response.get(self.ITEMS_KEY, ())
+        self._num_items = len(items)
+        self._remaining = self._num_items
+        self._item_iter = iter(items)
+
+    @property
+    def num_items(self):
+        """Total items in the page.
+
+        :rtype: int
+        :returns: The number of items in this page of items.
+        """
+        return self._num_items
+
+    @property
+    def remaining(self):
+        """Remaining items in the page.
+
+        :rtype: int
+        :returns: The number of items remaining this page.
+        """
+        return self._remaining
+
+    def __iter__(self):
+        """The :class:`Page` is an iterator."""
+        return self
+
+    def _item_to_value(self, item):
+        """Get the next item in the page.
+
+        This method (along with the constructor) is the workhorse
+        of this class. Subclasses will need to implement this method.
+
+        :type item: dict
+        :param item: An item to be converted to a native object.
+
+        :raises NotImplementedError: Always
+        """
+        raise NotImplementedError
+
+    def next(self):
+        """Get the next value in the iterator."""
+        item = six.next(self._item_iter)
+        result = self._item_to_value(item)
+        # Since we've successfully got the next value from the
+        # iterator, we update the number of remaining.
+        self._remaining -= 1
+        return result
+
+    # Alias needed for Python 2/3 support.
+    __next__ = next
+
+
 class Iterator(object):
     """A generic class for iterating through Cloud JSON APIs list responses.
 
+    Sub-classes need to over-write ``PAGE_CLASS``.
+
     :type client: :class:`google.cloud.client.Client`
     :param client: The client, which owns a connection to make requests.
 
-    :type path: str
-    :param path: The path to query for the list of items.
-
     :type page_token: str
     :param page_token: (Optional) A token identifying a page in a result set.
 
@@ -65,59 +158,74 @@ class Iterator(object):
 
     :type extra_params: dict or None
     :param extra_params: Extra query string parameters for the API call.
+
+    :type path: str
+    :param path: The path to query for the list of items.
     """
 
     PAGE_TOKEN = 'pageToken'
     MAX_RESULTS = 'maxResults'
     RESERVED_PARAMS = frozenset([PAGE_TOKEN, MAX_RESULTS])
+    PAGE_CLASS = Page
+    PATH = None
 
-    def __init__(self, client, path, page_token=None,
-                 max_results=None, extra_params=None):
+    def __init__(self, client, page_token=None, max_results=None,
+                 extra_params=None, path=None):
+        self.extra_params = extra_params or {}
+        self._verify_params()
+        self.max_results = max_results
         self.client = client
-        self.path = path
+        self.path = path or self.PATH
+        # The attributes below will change over the life of the iterator.
         self.page_number = 0
         self.next_page_token = page_token
-        self.max_results = max_results
         self.num_results = 0
-        self.extra_params = extra_params or {}
+        self._page = None
+
+    def _verify_params(self):
+        """Verifies the parameters don't use any reserved parameter.
+
+        :raises ValueError: If a reserved parameter is used.
+        """
         reserved_in_use = self.RESERVED_PARAMS.intersection(
             self.extra_params)
         if reserved_in_use:
-            raise ValueError(('Using a reserved parameter',
-                              reserved_in_use))
-        self._curr_items = iter(())
+            raise ValueError('Using a reserved parameter',
+                             reserved_in_use)
+
+    @property
+    def page(self):
+        """The current page of results that has been retrieved.
+
+        :rtype: :class:`Page`
+        :returns: The page of items that has been retrieved.
+        """
+        return self._page
 
     def __iter__(self):
         """The :class:`Iterator` is an iterator."""
         return self
 
-    def _update_items(self):
-        """Replace the current items iterator.
-
-        Intended to be used when the current items iterator is exhausted.
+    def _update_page(self):
+        """Replace the current page.
 
-        After replacing the iterator, consumes the first value to make sure
-        it is valid.
+        Does nothing if the current page is non-null and has items
+        remaining.
 
-        :rtype: object
-        :returns: The first item in the next iterator.
         :raises: :class:`~exceptions.StopIteration` if there is no next page.
         """
+        if self.page is not None and self.page.remaining > 0:
+            return
         if self.has_next_page():
-            response = self.get_next_page_response()
-            items = self.get_items_from_response(response)
-            self._curr_items = iter(items)
-            return six.next(self._curr_items)
+            response = self._get_next_page_response()
+            self._page = self.PAGE_CLASS(self, response)
         else:
             raise StopIteration
 
     def next(self):
         """Get the next value in the iterator."""
-        try:
-            item = six.next(self._curr_items)
-        except StopIteration:
-            item = self._update_items()
-
+        self._update_page()
+        item = six.next(self.page)
         self.num_results += 1
         return item
 
@@ -139,7 +247,7 @@ def has_next_page(self):
 
         return self.next_page_token is not None
 
-    def get_query_params(self):
+    def _get_query_params(self):
         """Getter for query parameters for the next request.
 
         :rtype: dict
@@ -153,17 +261,15 @@ def get_query_params(self):
         result.update(self.extra_params)
         return result
 
-    def get_next_page_response(self):
+    def _get_next_page_response(self):
         """Requests the next page from the path provided.
 
         :rtype: dict
         :returns: The parsed JSON response of the next page's contents.
         """
-        if not self.has_next_page():
-            raise RuntimeError('No more pages. Try resetting the iterator.')
-
         response = self.client.connection.api_request(
-            method='GET', path=self.path, query_params=self.get_query_params())
+            method='GET', path=self.path,
+            query_params=self._get_query_params())
 
         self.page_number += 1
         self.next_page_token = response.get('nextPageToken')
@@ -175,62 +281,4 @@ def reset(self):
         self.page_number = 0
         self.next_page_token = None
         self.num_results = 0
-
-    def get_items_from_response(self, response):
-        """Factory method called while iterating. This should be overridden.
-
-        This method should be overridden by a subclass.  It should
-        accept the API response of a request for the next page of items,
-        and return a list (or other iterable) of items.
-
-        Typically this method will construct a Bucket or a Blob from the
-        page of results in the response.
-
-        :type response: dict
-        :param response: The response of asking for the next page of items.
-        """
-        raise NotImplementedError
-
-
-class MethodIterator(object):
-    """Method-based iterator iterating through Cloud JSON APIs list responses.
-
-    :type method: instance method
-    :param method: ``list_foo`` method of a domain object, taking as arguments
-                   ``page_token``, ``page_size``, and optional additional
-                   keyword arguments.
-
-    :type page_token: string or ``NoneType``
-    :param page_token: Initial page token to pass.  if ``None``, fetch the
-                       first page from the ``method`` API call.
-
-    :type page_size: integer or ``NoneType``
-    :param page_size: Maximum number of items to return from the ``method``
-                      API call; if ``None``, uses the default for the API.
-
-    :type max_calls: integer or ``NoneType``
-    :param max_calls: Maximum number of times to make the ``method``
-                      API call; if ``None``, applies no limit.
-
-    :type kw: dict
-    :param kw: optional keyword arguments to be passed to ``method``.
-    """
-    def __init__(self, method, page_token=None, page_size=None,
-                 max_calls=None, **kw):
-        self._method = method
-        self._token = page_token
-        self._page_size = page_size
-        self._kw = kw
-        self._max_calls = max_calls
-        self._page_num = 0
-
-    def __iter__(self):
-        while self._max_calls is None or self._page_num < self._max_calls:
-            items, new_token = self._method(
-                page_token=self._token, page_size=self._page_size, **self._kw)
-            for item in items:
-                yield item
-            if new_token is None:
-                return
-            self._page_num += 1
-            self._token = new_token
+        self._page = None
diff --git a/core/unit_tests/test_iterator.py b/core/unit_tests/test_iterator.py
index 44d02d30770e..7f0a80b4e335 100644
--- a/core/unit_tests/test_iterator.py
+++ b/core/unit_tests/test_iterator.py
@@ -15,6 +15,76 @@
 import unittest
 
 
+class TestPage(unittest.TestCase):
+
+    def _getTargetClass(self):
+        from google.cloud.iterator import Page
+        return Page
+
+    def _makeOne(self, *args, **kw):
+        return self._getTargetClass()(*args, **kw)
+
+    def test_constructor(self):
+        klass = self._getTargetClass()
+        parent = object()
+        response = {klass.ITEMS_KEY: (1, 2, 3)}
+        page = self._makeOne(parent, response)
+        self.assertIs(page._parent, parent)
+        self.assertEqual(page._num_items, 3)
+        self.assertEqual(page._remaining, 3)
+
+    def test_num_items_property(self):
+        page = self._makeOne(None, {})
+        num_items = 42
+        page._num_items = num_items
+        self.assertEqual(page.num_items, num_items)
+
+    def test_remaining_property(self):
+        page = self._makeOne(None, {})
+        remaining = 1337
+        page._remaining = remaining
+        self.assertEqual(page.remaining, remaining)
+
+    def test___iter__(self):
+        page = self._makeOne(None, {})
+        self.assertIs(iter(page), page)
+
+    def test__item_to_value(self):
+        page = self._makeOne(None, {})
+        with self.assertRaises(NotImplementedError):
+            page._item_to_value(None)
+
+    def test_iterator_calls__item_to_value(self):
+        import six
+
+        klass = self._getTargetClass()
+
+        class CountItPage(klass):
+
+            calls = 0
+            values = None
+
+            def _item_to_value(self, item):
+                self.calls += 1
+                return item
+
+        response = {klass.ITEMS_KEY: [10, 11, 12]}
+        page = CountItPage(None, response)
+        page._remaining = 100
+
+        self.assertEqual(page.calls, 0)
+        self.assertEqual(page.remaining, 100)
+        self.assertEqual(six.next(page), 10)
+        self.assertEqual(page.calls, 1)
+        self.assertEqual(page.remaining, 99)
+        self.assertEqual(six.next(page), 11)
+        self.assertEqual(page.calls, 2)
+        self.assertEqual(page.remaining, 98)
+        self.assertEqual(six.next(page), 12)
+        self.assertEqual(page.calls, 3)
+        self.assertEqual(page.remaining, 97)
+
+
 class TestIterator(unittest.TestCase):
 
     def _getTargetClass(self):
@@ -24,23 +94,37 @@ def _getTargetClass(self):
     def _makeOne(self, *args, **kw):
         return self._getTargetClass()(*args, **kw)
 
-    def test_ctor(self):
+    def test_constructor(self):
+        connection = _Connection()
+        client = _Client(connection)
+        path = '/foo'
+        iterator = self._makeOne(client, path=path)
+        self.assertIs(iterator.client, client)
+        self.assertEqual(iterator.path, path)
+        self.assertEqual(iterator.page_number, 0)
+        self.assertIsNone(iterator.next_page_token)
+
+    def test_constructor_default_path(self):
+        klass = self._getTargetClass()
+
+        class WithPath(klass):
+            PATH = '/path'
+
         connection = _Connection()
         client = _Client(connection)
-        PATH = '/foo'
-        iterator = self._makeOne(client, PATH)
+        iterator = WithPath(client)
         self.assertIs(iterator.client, client)
-        self.assertEqual(iterator.path, PATH)
+        self.assertEqual(iterator.path, WithPath.PATH)
         self.assertEqual(iterator.page_number, 0)
         self.assertIsNone(iterator.next_page_token)
 
     def test_constructor_w_extra_param_collision(self):
         connection = _Connection()
         client = _Client(connection)
-        PATH = '/foo'
+        path = '/foo'
         extra_params = {'pageToken': 'val'}
-        self.assertRaises(ValueError, self._makeOne, client, PATH,
-                          extra_params=extra_params)
+        with self.assertRaises(ValueError):
+            self._makeOne(client, path=path, extra_params=extra_params)
 
     def test___iter__(self):
         iterator = self._makeOne(None, None)
@@ -48,30 +132,32 @@ def test___iter__(self):
 
     def test_iterate(self):
         import six
+        from google.cloud.iterator import Page
 
-        PATH = '/foo'
-        KEY1 = 'key1'
-        KEY2 = 'key2'
-        ITEM1, ITEM2 = object(), object()
-        ITEMS = {KEY1: ITEM1, KEY2: ITEM2}
+        path = '/foo'
+        key1 = 'key1'
+        key2 = 'key2'
+        item1, item2 = object(), object()
+        ITEMS = {key1: item1, key2: item2}
+
+        class _Page(Page):
 
-        def _get_items(response):
-            return [ITEMS[item['name']]
-                    for item in response.get('items', [])]
+            def _item_to_value(self, item):
+                return ITEMS[item['name']]
 
         connection = _Connection(
-            {'items': [{'name': KEY1}, {'name': KEY2}]})
+            {'items': [{'name': key1}, {'name': key2}]})
         client = _Client(connection)
-        iterator = self._makeOne(client, PATH)
-        iterator.get_items_from_response = _get_items
+        iterator = self._makeOne(client, path=path)
+        iterator.PAGE_CLASS = _Page
         self.assertEqual(iterator.num_results, 0)
 
         val1 = six.next(iterator)
-        self.assertEqual(val1, ITEM1)
+        self.assertEqual(val1, item1)
         self.assertEqual(iterator.num_results, 1)
 
         val2 = six.next(iterator)
-        self.assertEqual(val2, ITEM2)
+        self.assertEqual(val2, item2)
         self.assertEqual(iterator.num_results, 2)
 
         with self.assertRaises(StopIteration):
@@ -79,36 +165,36 @@ def _get_items(response):
 
         kw, = connection._requested
         self.assertEqual(kw['method'], 'GET')
-        self.assertEqual(kw['path'], PATH)
+        self.assertEqual(kw['path'], path)
         self.assertEqual(kw['query_params'], {})
 
     def test_has_next_page_new(self):
         connection = _Connection()
         client = _Client(connection)
-        PATH = '/foo'
-        iterator = self._makeOne(client, PATH)
+        path = '/foo'
+        iterator = self._makeOne(client, path=path)
         self.assertTrue(iterator.has_next_page())
 
     def test_has_next_page_w_number_no_token(self):
         connection = _Connection()
         client = _Client(connection)
-        PATH = '/foo'
-        iterator = self._makeOne(client, PATH)
+        path = '/foo'
+        iterator = self._makeOne(client, path=path)
         iterator.page_number = 1
         self.assertFalse(iterator.has_next_page())
 
     def test_has_next_page_w_number_w_token(self):
         connection = _Connection()
         client = _Client(connection)
-        PATH = '/foo'
-        TOKEN = 'token'
-        iterator = self._makeOne(client, PATH)
+        path = '/foo'
+        token = 'token'
+        iterator = self._makeOne(client, path=path)
         iterator.page_number = 1
-        iterator.next_page_token = TOKEN
+        iterator.next_page_token = token
         self.assertTrue(iterator.has_next_page())
 
     def test_has_next_page_w_max_results_not_done(self):
-        iterator = self._makeOne(None, None, max_results=3,
+        iterator = self._makeOne(None, path=None, max_results=3,
                                  page_token='definitely-not-none')
         iterator.page_number = 1
         self.assertLess(iterator.num_results, iterator.max_results)
@@ -120,189 +206,88 @@ def test_has_next_page_w_max_results_done(self):
         iterator.num_results = iterator.max_results
         self.assertFalse(iterator.has_next_page())
 
-    def test_get_query_params_no_token(self):
+    def test__get_query_params_no_token(self):
         connection = _Connection()
         client = _Client(connection)
-        PATH = '/foo'
-        iterator = self._makeOne(client, PATH)
-        self.assertEqual(iterator.get_query_params(), {})
+        path = '/foo'
+        iterator = self._makeOne(client, path=path)
+        self.assertEqual(iterator._get_query_params(), {})
 
-    def test_get_query_params_w_token(self):
+    def test__get_query_params_w_token(self):
         connection = _Connection()
         client = _Client(connection)
-        PATH = '/foo'
-        TOKEN = 'token'
-        iterator = self._makeOne(client, PATH)
-        iterator.next_page_token = TOKEN
-        self.assertEqual(iterator.get_query_params(),
-                         {'pageToken': TOKEN})
-
-    def test_get_query_params_w_max_results(self):
+        path = '/foo'
+        token = 'token'
+        iterator = self._makeOne(client, path=path)
+        iterator.next_page_token = token
+        self.assertEqual(iterator._get_query_params(),
+                         {'pageToken': token})
+
+    def test__get_query_params_w_max_results(self):
         connection = _Connection()
         client = _Client(connection)
         path = '/foo'
         max_results = 3
-        iterator = self._makeOne(client, path,
+        iterator = self._makeOne(client, path=path,
                                  max_results=max_results)
         iterator.num_results = 1
         local_max = max_results - iterator.num_results
-        self.assertEqual(iterator.get_query_params(),
+        self.assertEqual(iterator._get_query_params(),
                          {'maxResults': local_max})
 
-    def test_get_query_params_extra_params(self):
+    def test__get_query_params_extra_params(self):
         connection = _Connection()
         client = _Client(connection)
-        PATH = '/foo'
+        path = '/foo'
         extra_params = {'key': 'val'}
-        iterator = self._makeOne(client, PATH, extra_params=extra_params)
-        self.assertEqual(iterator.get_query_params(), extra_params)
+        iterator = self._makeOne(client, path=path, extra_params=extra_params)
+        self.assertEqual(iterator._get_query_params(), extra_params)
 
-    def test_get_query_params_w_token_and_extra_params(self):
+    def test__get_query_params_w_token_and_extra_params(self):
         connection = _Connection()
         client = _Client(connection)
-        PATH = '/foo'
-        TOKEN = 'token'
+        path = '/foo'
+        token = 'token'
         extra_params = {'key': 'val'}
-        iterator = self._makeOne(client, PATH, extra_params=extra_params)
-        iterator.next_page_token = TOKEN
+        iterator = self._makeOne(client, path=path, extra_params=extra_params)
+        iterator.next_page_token = token
 
         expected_query = extra_params.copy()
-        expected_query.update({'pageToken': TOKEN})
-        self.assertEqual(iterator.get_query_params(), expected_query)
-
-    def test_get_next_page_response_new_no_token_in_response(self):
-        PATH = '/foo'
-        TOKEN = 'token'
-        KEY1 = 'key1'
-        KEY2 = 'key2'
-        connection = _Connection({'items': [{'name': KEY1}, {'name': KEY2}],
-                                  'nextPageToken': TOKEN})
+        expected_query.update({'pageToken': token})
+        self.assertEqual(iterator._get_query_params(), expected_query)
+
+    def test__get_next_page_response_new_no_token_in_response(self):
+        path = '/foo'
+        token = 'token'
+        key1 = 'key1'
+        key2 = 'key2'
+        connection = _Connection({'items': [{'name': key1}, {'name': key2}],
+                                  'nextPageToken': token})
         client = _Client(connection)
-        iterator = self._makeOne(client, PATH)
-        response = iterator.get_next_page_response()
-        self.assertEqual(response['items'], [{'name': KEY1}, {'name': KEY2}])
+        iterator = self._makeOne(client, path=path)
+        response = iterator._get_next_page_response()
+        self.assertEqual(response['items'], [{'name': key1}, {'name': key2}])
         self.assertEqual(iterator.page_number, 1)
-        self.assertEqual(iterator.next_page_token, TOKEN)
+        self.assertEqual(iterator.next_page_token, token)
         kw, = connection._requested
         self.assertEqual(kw['method'], 'GET')
-        self.assertEqual(kw['path'], PATH)
+        self.assertEqual(kw['path'], path)
         self.assertEqual(kw['query_params'], {})
 
-    def test_get_next_page_response_no_token(self):
-        connection = _Connection()
-        client = _Client(connection)
-        PATH = '/foo'
-        iterator = self._makeOne(client, PATH)
-        iterator.page_number = 1
-        self.assertRaises(RuntimeError, iterator.get_next_page_response)
-
     def test_reset(self):
         connection = _Connection()
         client = _Client(connection)
-        PATH = '/foo'
-        TOKEN = 'token'
-        iterator = self._makeOne(client, PATH)
+        path = '/foo'
+        token = 'token'
+        iterator = self._makeOne(client, path=path)
         iterator.page_number = 1
-        iterator.next_page_token = TOKEN
+        iterator.next_page_token = token
+        iterator._page = object()
         iterator.reset()
         self.assertEqual(iterator.page_number, 0)
+        self.assertEqual(iterator.num_results, 0)
         self.assertIsNone(iterator.next_page_token)
-
-    def test_get_items_from_response_raises_NotImplementedError(self):
-        PATH = '/foo'
-        connection = _Connection()
-        client = _Client(connection)
-        iterator = self._makeOne(client, PATH)
-        self.assertRaises(NotImplementedError,
-                          iterator.get_items_from_response, object())
-
-
-class TestMethodIterator(unittest.TestCase):
-
-    def _getTargetClass(self):
-        from google.cloud.iterator import MethodIterator
-        return MethodIterator
-
-    def _makeOne(self, *args, **kw):
-        return self._getTargetClass()(*args, **kw)
-
-    def test_ctor_defaults(self):
-        wlm = _WithListMethod()
-        iterator = self._makeOne(wlm.list_foo)
-        self.assertEqual(iterator._method, wlm.list_foo)
-        self.assertIsNone(iterator._token)
-        self.assertIsNone(iterator._page_size)
-        self.assertEqual(iterator._kw, {})
-        self.assertIsNone(iterator._max_calls)
-        self.assertEqual(iterator._page_num, 0)
-
-    def test_ctor_explicit(self):
-        wlm = _WithListMethod()
-        TOKEN = wlm._letters
-        SIZE = 4
-        CALLS = 2
-        iterator = self._makeOne(wlm.list_foo, TOKEN, SIZE, CALLS,
-                                 foo_type='Bar')
-        self.assertEqual(iterator._method, wlm.list_foo)
-        self.assertEqual(iterator._token, TOKEN)
-        self.assertEqual(iterator._page_size, SIZE)
-        self.assertEqual(iterator._kw, {'foo_type': 'Bar'})
-        self.assertEqual(iterator._max_calls, CALLS)
-        self.assertEqual(iterator._page_num, 0)
-
-    def test___iter___defaults(self):
-        import string
-        wlm = _WithListMethod()
-        iterator = self._makeOne(wlm.list_foo)
-        found = []
-        for char in iterator:
-            found.append(char)
-        self.assertEqual(found, list(string.printable))
-        self.assertEqual(len(wlm._called_with), len(found) // 10)
-        for i, (token, size, kw) in enumerate(wlm._called_with):
-            if i == 0:
-                self.assertIsNone(token)
-            else:
-                self.assertEqual(token, string.printable[i * 10:])
-            self.assertIsNone(size)
-            self.assertEqual(kw, {})
-
-    def test___iter___explicit_size_and_maxcalls_and_kw(self):
-        import string
-        wlm = _WithListMethod()
-        iterator = self._makeOne(wlm.list_foo, page_size=2, max_calls=3,
-                                 foo_type='Bar')
-        found = []
-        for char in iterator:
-            found.append(char)
-        self.assertEqual(found, list(string.printable[:2 * 3]))
-        self.assertEqual(len(wlm._called_with), len(found) // 2)
-        for i, (token, size, kw) in enumerate(wlm._called_with):
-            if i == 0:
-                self.assertIsNone(token)
-            else:
-                self.assertEqual(token, string.printable[i * 2:])
-            self.assertEqual(size, 2)
-            self.assertEqual(kw, {'foo_type': 'Bar'})
-
-
-class _WithListMethod(object):
-
-    def __init__(self):
-        import string
-        self._called_with = []
-        self._letters = string.printable
-
-    def list_foo(self, page_token, page_size, **kw):
-        if page_token is not None:
-            assert page_token == self._letters
-        self._called_with.append((page_token, page_size, kw))
-        if page_size is None:
-            page_size = 10
-        page, self._letters = (
-            self._letters[:page_size], self._letters[page_size:])
-        token = self._letters or None
-        return page, token
+        self.assertIsNone(iterator.page)
 
 
 class _Connection(object):
diff --git a/docs/google-cloud-api.rst b/docs/google-cloud-api.rst
index 0fb79d966cfb..c95d9ae2bce8 100644
--- a/docs/google-cloud-api.rst
+++ b/docs/google-cloud-api.rst
@@ -36,3 +36,10 @@ Environment Variables
 .. automodule:: google.cloud.environment_vars
   :members:
   :show-inheritance:
+
+Base Iterator Class
+~~~~~~~~~~~~~~~~~~~
+
+.. automodule:: google.cloud.iterator
+  :members:
+  :show-inheritance:
diff --git a/resource_manager/google/cloud/resource_manager/client.py b/resource_manager/google/cloud/resource_manager/client.py
index 80d7392bb9f6..1f205edb6fc9 100644
--- a/resource_manager/google/cloud/resource_manager/client.py
+++ b/resource_manager/google/cloud/resource_manager/client.py
@@ -17,6 +17,7 @@
 
 from google.cloud.client import Client as BaseClient
 from google.cloud.iterator import Iterator
+from google.cloud.iterator import Page
 from google.cloud.resource_manager.connection import Connection
 from google.cloud.resource_manager.project import Project
 
@@ -158,6 +159,30 @@ def list_projects(self, filter_params=None, page_size=None):
         return _ProjectIterator(self, extra_params=extra_params)
 
 
+class _ProjectPage(Page):
+    """Iterator for a single page of results.
+
+    :type parent: :class:`_ProjectIterator`
+    :param parent: The iterator that owns the current page.
+
+    :type response: dict
+    :param response: The JSON API response for a page of projects.
+    """
+
+    ITEMS_KEY = 'projects'
+
+    def _item_to_value(self, resource):
+        """Convert a JSON project to the native object.
+
+        :type resource: dict
+        :param resource: An resource to be converted to a project.
+
+        :rtype: :class:`.Project`
+        :returns: The next project in the page.
+        """
+        return Project.from_api_repr(resource, client=self._parent.client)
+
+
 class _ProjectIterator(Iterator):
     """An iterator over a list of Project resources.
 
@@ -179,18 +204,5 @@ class _ProjectIterator(Iterator):
                          the API call.
     """
 
-    def __init__(self, client, page_token=None,
-                 max_results=None, extra_params=None):
-        super(_ProjectIterator, self).__init__(
-            client=client, path='/projects', page_token=page_token,
-            max_results=max_results, extra_params=extra_params)
-
-    def get_items_from_response(self, response):
-        """Yield projects from response.
-
-        :type response: dict
-        :param response: The JSON API response for a page of projects.
-        """
-        for resource in response.get('projects', []):
-            item = Project.from_api_repr(resource, client=self.client)
-            yield item
+    PAGE_CLASS = _ProjectPage
+    PATH = '/projects'
diff --git a/resource_manager/unit_tests/test_client.py b/resource_manager/unit_tests/test_client.py
index 71eb717aee9c..fbfbeb4c7c63 100644
--- a/resource_manager/unit_tests/test_client.py
+++ b/resource_manager/unit_tests/test_client.py
@@ -15,58 +15,76 @@
 import unittest
 
 
-class Test__ProjectIterator(unittest.TestCase):
+class Test__ProjectPage(unittest.TestCase):
 
     def _getTargetClass(self):
-        from google.cloud.resource_manager.client import _ProjectIterator
-        return _ProjectIterator
+        from google.cloud.resource_manager.client import _ProjectPage
+        return _ProjectPage
 
     def _makeOne(self, *args, **kw):
         return self._getTargetClass()(*args, **kw)
 
-    def test_constructor(self):
-        client = object()
-        iterator = self._makeOne(client)
-        self.assertEqual(iterator.path, '/projects')
-        self.assertEqual(iterator.page_number, 0)
-        self.assertIsNone(iterator.next_page_token)
-        self.assertIs(iterator.client, client)
-        self.assertEqual(iterator.extra_params, {})
+    def test_empty_response(self):
+        from google.cloud.resource_manager.client import _ProjectIterator
 
-    def test_get_items_from_response_empty(self):
         client = object()
-        iterator = self._makeOne(client)
-        self.assertEqual(list(iterator.get_items_from_response({})), [])
+        iterator = _ProjectIterator(client)
+        page = self._makeOne(iterator, {})
+        self.assertEqual(page.num_items, 0)
+        self.assertEqual(page.remaining, 0)
+        self.assertEqual(list(page), [])
 
-    def test_get_items_from_response_non_empty(self):
+    def test_non_empty_response(self):
+        from google.cloud.resource_manager.client import _ProjectIterator
         from google.cloud.resource_manager.project import Project
 
-        PROJECT_ID = 'project-id'
-        PROJECT_NAME = 'My Project Name'
-        PROJECT_NUMBER = 12345678
-        PROJECT_LABELS = {'env': 'prod'}
-        PROJECT_LIFECYCLE_STATE = 'ACTIVE'
-        API_RESOURCE = {
-            'projectId': PROJECT_ID,
-            'name': PROJECT_NAME,
-            'projectNumber': PROJECT_NUMBER,
-            'labels': PROJECT_LABELS,
-            'lifecycleState': PROJECT_LIFECYCLE_STATE,
+        project_id = 'project-id'
+        project_name = 'My Project Name'
+        project_number = 12345678
+        project_labels = {'env': 'prod'}
+        project_lifecycle_state = 'ACTIVE'
+        api_resource = {
+            'projectId': project_id,
+            'name': project_name,
+            'projectNumber': project_number,
+            'labels': project_labels,
+            'lifecycleState': project_lifecycle_state,
         }
-        RESPONSE = {'projects': [API_RESOURCE]}
+        response = {'projects': [api_resource]}
 
         client = object()
-        iterator = self._makeOne(client)
-        projects = list(iterator.get_items_from_response(RESPONSE))
+        iterator = _ProjectIterator(client)
+        page = self._makeOne(iterator, response)
 
-        project, = projects
+        self.assertEqual(page.num_items, 1)
+        project = page.next()
+        self.assertEqual(page.remaining, 0)
         self.assertIsInstance(project, Project)
-        self.assertEqual(project.project_id, PROJECT_ID)
+        self.assertEqual(project.project_id, project_id)
         self.assertEqual(project._client, client)
-        self.assertEqual(project.name, PROJECT_NAME)
-        self.assertEqual(project.number, PROJECT_NUMBER)
-        self.assertEqual(project.labels, PROJECT_LABELS)
-        self.assertEqual(project.status, PROJECT_LIFECYCLE_STATE)
+        self.assertEqual(project.name, project_name)
+        self.assertEqual(project.number, project_number)
+        self.assertEqual(project.labels, project_labels)
+        self.assertEqual(project.status, project_lifecycle_state)
+
+
+class Test__ProjectIterator(unittest.TestCase):
+
+    def _getTargetClass(self):
+        from google.cloud.resource_manager.client import _ProjectIterator
+        return _ProjectIterator
+
+    def _makeOne(self, *args, **kw):
+        return self._getTargetClass()(*args, **kw)
+
+    def test_constructor(self):
+        client = object()
+        iterator = self._makeOne(client)
+        self.assertEqual(iterator.path, '/projects')
+        self.assertEqual(iterator.page_number, 0)
+        self.assertIsNone(iterator.next_page_token)
+        self.assertIs(iterator.client, client)
+        self.assertEqual(iterator.extra_params, {})
 
 
 class TestClient(unittest.TestCase):
diff --git a/scripts/verify_included_modules.py b/scripts/verify_included_modules.py
index ed447585e2d5..d0791f0807b0 100644
--- a/scripts/verify_included_modules.py
+++ b/scripts/verify_included_modules.py
@@ -36,7 +36,6 @@
     'google.cloud.datastore.__init__',
     'google.cloud.dns.__init__',
     'google.cloud.error_reporting.__init__',
-    'google.cloud.iterator',
     'google.cloud.language.__init__',
     'google.cloud.logging.__init__',
     'google.cloud.logging.handlers.__init__',
diff --git a/storage/google/cloud/storage/bucket.py b/storage/google/cloud/storage/bucket.py
index 3068bfb2fe06..c8e56a4dbcf1 100644
--- a/storage/google/cloud/storage/bucket.py
+++ b/storage/google/cloud/storage/bucket.py
@@ -21,6 +21,7 @@
 from google.cloud._helpers import _rfc3339_to_datetime
 from google.cloud.exceptions import NotFound
 from google.cloud.iterator import Iterator
+from google.cloud.iterator import Page
 from google.cloud.storage._helpers import _PropertyMixin
 from google.cloud.storage._helpers import _scalar_property
 from google.cloud.storage.acl import BucketACL
@@ -28,6 +29,37 @@
 from google.cloud.storage.blob import Blob
 
 
+class _BlobPage(Page):
+    """Iterator for a single page of results.
+
+    :type parent: :class:`_BlobIterator`
+    :param parent: The iterator that owns the current page.
+
+    :type response: dict
+    :param response: The JSON API response for a page of blobs.
+    """
+
+    def __init__(self, parent, response):
+        super(_BlobPage, self).__init__(parent, response)
+        # Grab the prefixes from the response.
+        self._prefixes = tuple(response.get('prefixes', ()))
+        parent.prefixes.update(self._prefixes)
+
+    def _item_to_value(self, item):
+        """Convert a JSON blob to the native object.
+
+        :type item: dict
+        :param item: An item to be converted to a blob.
+
+        :rtype: :class:`.Blob`
+        :returns: The next blob in the page.
+        """
+        name = item.get('name')
+        blob = Blob(name, bucket=self._parent.bucket)
+        blob._set_properties(item)
+        return blob
+
+
 class _BlobIterator(Iterator):
     """An iterator listing blobs in a bucket
 
@@ -50,32 +82,20 @@ class _BlobIterator(Iterator):
     :param client: Optional. The client to use for making connections.
                    Defaults to the bucket's client.
     """
+
+    PAGE_CLASS = _BlobPage
+
     def __init__(self, bucket, page_token=None, max_results=None,
                  extra_params=None, client=None):
         if client is None:
             client = bucket.client
         self.bucket = bucket
         self.prefixes = set()
-        self._current_prefixes = None
         super(_BlobIterator, self).__init__(
             client=client, path=bucket.path + '/o',
             page_token=page_token, max_results=max_results,
             extra_params=extra_params)
 
-    def get_items_from_response(self, response):
-        """Yield :class:`.storage.blob.Blob` items from response.
-
-        :type response: dict
-        :param response: The JSON API response for a page of blobs.
-        """
-        self._current_prefixes = tuple(response.get('prefixes', ()))
-        self.prefixes.update(self._current_prefixes)
-        for item in response.get('items', []):
-            name = item.get('name')
-            blob = Blob(name, bucket=self.bucket)
-            blob._set_properties(item)
-            yield blob
-
 
 class Bucket(_PropertyMixin):
     """A class representing a Bucket on Cloud Storage.
diff --git a/storage/google/cloud/storage/client.py b/storage/google/cloud/storage/client.py
index c5eb22158e07..94e0d9406ffa 100644
--- a/storage/google/cloud/storage/client.py
+++ b/storage/google/cloud/storage/client.py
@@ -19,6 +19,7 @@
 from google.cloud.client import JSONClient
 from google.cloud.exceptions import NotFound
 from google.cloud.iterator import Iterator
+from google.cloud.iterator import Page
 from google.cloud.storage.batch import Batch
 from google.cloud.storage.bucket import Bucket
 from google.cloud.storage.connection import Connection
@@ -271,6 +272,31 @@ def list_buckets(self, max_results=None, page_token=None, prefix=None,
         return result
 
 
+class _BucketPage(Page):
+    """Iterator for a single page of results.
+
+    :type parent: :class:`_BucketIterator`
+    :param parent: The iterator that owns the current page.
+
+    :type response: dict
+    :param response: The JSON API response for a page of buckets.
+    """
+
+    def _item_to_value(self, item):
+        """Convert a JSON bucket to the native object.
+
+        :type item: dict
+        :param item: An item to be converted to a bucket.
+
+        :rtype: :class:`.Bucket`
+        :returns: The next bucket in the page.
+        """
+        name = item.get('name')
+        bucket = Bucket(self._parent.client, name)
+        bucket._set_properties(item)
+        return bucket
+
+
 class _BucketIterator(Iterator):
     """An iterator listing all buckets.
 
@@ -291,21 +317,5 @@ class _BucketIterator(Iterator):
     :param extra_params: Extra query string parameters for the API call.
     """
 
-    def __init__(self, client, page_token=None,
-                 max_results=None, extra_params=None):
-        super(_BucketIterator, self).__init__(
-            client=client, path='/b',
-            page_token=page_token, max_results=max_results,
-            extra_params=extra_params)
-
-    def get_items_from_response(self, response):
-        """Factory method which yields :class:`.Bucket` items from a response.
-
-        :type response: dict
-        :param response: The JSON API response for a page of buckets.
-        """
-        for item in response.get('items', []):
-            name = item.get('name')
-            bucket = Bucket(self.client, name)
-            bucket._set_properties(item)
-            yield bucket
+    PAGE_CLASS = _BucketPage
+    PATH = '/b'
diff --git a/storage/unit_tests/test_bucket.py b/storage/unit_tests/test_bucket.py
index a0b8c1847b1d..924789a102ca 100644
--- a/storage/unit_tests/test_bucket.py
+++ b/storage/unit_tests/test_bucket.py
@@ -15,6 +15,48 @@
 import unittest
 
 
+class Test__BlobPage(unittest.TestCase):
+
+    def _getTargetClass(self):
+        from google.cloud.storage.bucket import _BlobPage
+        return _BlobPage
+
+    def _makeOne(self, *args, **kw):
+        return self._getTargetClass()(*args, **kw)
+
+    def test_empty_response(self):
+        from google.cloud.storage.bucket import _BlobIterator
+
+        connection = _Connection()
+        client = _Client(connection)
+        bucket = _Bucket()
+        iterator = _BlobIterator(bucket, client=client)
+        page = self._makeOne(iterator, {})
+        blobs = list(page)
+        self.assertEqual(blobs, [])
+        self.assertEqual(iterator.prefixes, set())
+
+    def test_non_empty_response(self):
+        from google.cloud.storage.blob import Blob
+        from google.cloud.storage.bucket import _BlobIterator
+
+        blob_name = 'blob-name'
+        response = {'items': [{'name': blob_name}], 'prefixes': ['foo']}
+        connection = _Connection()
+        client = _Client(connection)
+        bucket = _Bucket()
+        iterator = _BlobIterator(bucket, client=client)
+        page = self._makeOne(iterator, response)
+
+        self.assertEqual(page._prefixes, ('foo',))
+        self.assertEqual(page.num_items, 1)
+        blob = page.next()
+        self.assertEqual(page.remaining, 0)
+        self.assertIsInstance(blob, Blob)
+        self.assertEqual(blob.name, blob_name)
+        self.assertEqual(iterator.prefixes, set(['foo']))
+
+
 class Test__BlobIterator(unittest.TestCase):
 
     def _getTargetClass(self):
@@ -36,52 +78,36 @@ def test_ctor(self):
         self.assertIsNone(iterator.next_page_token)
         self.assertEqual(iterator.prefixes, set())
 
-    def test_get_items_from_response_empty(self):
-        connection = _Connection()
-        client = _Client(connection)
-        bucket = _Bucket()
-        iterator = self._makeOne(bucket, client=client)
-        blobs = list(iterator.get_items_from_response({}))
-        self.assertEqual(blobs, [])
-        self.assertEqual(iterator.prefixes, set())
-
-    def test_get_items_from_response_non_empty(self):
+    def test_cumulative_prefixes(self):
         from google.cloud.storage.blob import Blob
-        BLOB_NAME = 'blob-name'
-        response = {'items': [{'name': BLOB_NAME}], 'prefixes': ['foo']}
-        connection = _Connection()
-        client = _Client(connection)
-        bucket = _Bucket()
-        iterator = self._makeOne(bucket, client=client)
-        blobs = list(iterator.get_items_from_response(response))
-        self.assertEqual(len(blobs), 1)
-        blob = blobs[0]
-        self.assertIsInstance(blob, Blob)
-        self.assertEqual(blob.name, BLOB_NAME)
-        self.assertEqual(iterator.prefixes, set(['foo']))
+        from google.cloud.storage.bucket import _BlobPage
 
-    def test_get_items_from_response_cumulative_prefixes(self):
-        from google.cloud.storage.blob import Blob
         BLOB_NAME = 'blob-name1'
-        response1 = {'items': [{'name': BLOB_NAME}], 'prefixes': ['foo']}
+        response1 = {
+            'items': [{'name': BLOB_NAME}],
+            'prefixes': ['foo'],
+        }
         response2 = {
             'items': [],
-            'prefixes': ['foo', 'bar'],
+            'prefixes': ['bar'],
         }
         connection = _Connection()
         client = _Client(connection)
         bucket = _Bucket()
         iterator = self._makeOne(bucket, client=client)
         # Parse first response.
-        blobs = list(iterator.get_items_from_response(response1))
-        self.assertEqual(len(blobs), 1)
-        blob = blobs[0]
+        page1 = _BlobPage(iterator, response1)
+        self.assertEqual(page1._prefixes, ('foo',))
+        self.assertEqual(page1.num_items, 1)
+        blob = page1.next()
+        self.assertEqual(page1.remaining, 0)
         self.assertIsInstance(blob, Blob)
         self.assertEqual(blob.name, BLOB_NAME)
         self.assertEqual(iterator.prefixes, set(['foo']))
         # Parse second response.
-        blobs = list(iterator.get_items_from_response(response2))
-        self.assertEqual(len(blobs), 0)
+        page2 = _BlobPage(iterator, response2)
+        self.assertEqual(page2._prefixes, ('bar',))
+        self.assertEqual(page2.num_items, 0)
         self.assertEqual(iterator.prefixes, set(['foo', 'bar']))
 
 
@@ -944,6 +970,8 @@ def test_make_public_w_future_reload_default(self):
     def test_make_public_recursive(self):
         from google.cloud.storage.acl import _ACLEntity
         from google.cloud.storage.bucket import _BlobIterator
+        from google.cloud.storage.bucket import _BlobPage
+
         _saved = []
 
         class _Blob(object):
@@ -968,10 +996,12 @@ def save(self, client=None):
                 _saved.append(
                     (self._bucket, self._name, self._granted, client))
 
+        class _Page(_BlobPage):
+            def _item_to_value(self, item):
+                return _Blob(self._parent.bucket, item['name'])
+
         class _Iterator(_BlobIterator):
-            def get_items_from_response(self, response):
-                for item in response.get('items', []):
-                    yield _Blob(self.bucket, item['name'])
+            PAGE_CLASS = _Page
 
         NAME = 'name'
         BLOB_NAME = 'blob-name'
diff --git a/storage/unit_tests/test_client.py b/storage/unit_tests/test_client.py
index 1ee5e139a810..f75965c268fb 100644
--- a/storage/unit_tests/test_client.py
+++ b/storage/unit_tests/test_client.py
@@ -370,44 +370,60 @@ def test_list_buckets_all_arguments(self):
         self.assertEqual(parse_qs(uri_parts.query), EXPECTED_QUERY)
 
 
-class Test__BucketIterator(unittest.TestCase):
+class Test__BucketPage(unittest.TestCase):
 
     def _getTargetClass(self):
-        from google.cloud.storage.client import _BucketIterator
-        return _BucketIterator
+        from google.cloud.storage.client import _BucketPage
+        return _BucketPage
 
     def _makeOne(self, *args, **kw):
         return self._getTargetClass()(*args, **kw)
 
-    def test_ctor(self):
-        connection = object()
-        client = _Client(connection)
-        iterator = self._makeOne(client)
-        self.assertEqual(iterator.path, '/b')
-        self.assertEqual(iterator.page_number, 0)
-        self.assertIsNone(iterator.next_page_token)
-        self.assertIs(iterator.client, client)
+    def test_empty_response(self):
+        from google.cloud.storage.client import _BucketIterator
 
-    def test_get_items_from_response_empty(self):
         connection = object()
         client = _Client(connection)
-        iterator = self._makeOne(client)
-        self.assertEqual(list(iterator.get_items_from_response({})), [])
+        iterator = _BucketIterator(client)
+        page = self._makeOne(iterator, {})
+        self.assertEqual(list(page), [])
 
-    def test_get_items_from_response_non_empty(self):
+    def test_non_empty_response(self):
         from google.cloud.storage.bucket import Bucket
+        from google.cloud.storage.client import _BucketIterator
+
         BLOB_NAME = 'blob-name'
         response = {'items': [{'name': BLOB_NAME}]}
         connection = object()
         client = _Client(connection)
-        iterator = self._makeOne(client)
-        buckets = list(iterator.get_items_from_response(response))
-        self.assertEqual(len(buckets), 1)
-        bucket = buckets[0]
+        iterator = _BucketIterator(client)
+        page = self._makeOne(iterator, response)
+        self.assertEqual(page.num_items, 1)
+        bucket = page.next()
+        self.assertEqual(page.remaining, 0)
         self.assertIsInstance(bucket, Bucket)
         self.assertEqual(bucket.name, BLOB_NAME)
 
 
+class Test__BucketIterator(unittest.TestCase):
+
+    def _getTargetClass(self):
+        from google.cloud.storage.client import _BucketIterator
+        return _BucketIterator
+
+    def _makeOne(self, *args, **kw):
+        return self._getTargetClass()(*args, **kw)
+
+    def test_ctor(self):
+        connection = object()
+        client = _Client(connection)
+        iterator = self._makeOne(client)
+        self.assertEqual(iterator.path, '/b')
+        self.assertEqual(iterator.page_number, 0)
+        self.assertIsNone(iterator.next_page_token)
+        self.assertIs(iterator.client, client)
+
+
 class _Credentials(object):
 
     _scopes = None
diff --git a/system_tests/storage.py b/system_tests/storage.py
index 664cab102d0b..75faf6faf633 100644
--- a/system_tests/storage.py
+++ b/system_tests/storage.py
@@ -257,14 +257,13 @@ def test_paginate_files(self):
         truncation_size = 1
         count = len(self.FILENAMES) - truncation_size
         iterator = self.bucket.list_blobs(max_results=count)
-        response = iterator.get_next_page_response()
-        blobs = list(iterator.get_items_from_response(response))
+        iterator._update_page()
+        blobs = list(iterator.page)
         self.assertEqual(len(blobs), count)
-        self.assertEqual(iterator.page_number, 1)
         self.assertIsNotNone(iterator.next_page_token)
 
-        response = iterator.get_next_page_response()
-        last_blobs = list(iterator.get_items_from_response(response))
+        iterator._update_page()
+        last_blobs = list(iterator.page)
         self.assertEqual(len(last_blobs), truncation_size)
 
 
@@ -302,20 +301,18 @@ def tearDownClass(cls):
     @RetryErrors(unittest.TestCase.failureException)
     def test_root_level_w_delimiter(self):
         iterator = self.bucket.list_blobs(delimiter='/')
-        response = iterator.get_next_page_response()
-        blobs = list(iterator.get_items_from_response(response))
+        iterator._update_page()
+        blobs = list(iterator.page)
         self.assertEqual([blob.name for blob in blobs], ['file01.txt'])
-        self.assertEqual(iterator.page_number, 1)
         self.assertIsNone(iterator.next_page_token)
         self.assertEqual(iterator.prefixes, set(['parent/']))
 
     @RetryErrors(unittest.TestCase.failureException)
     def test_first_level(self):
         iterator = self.bucket.list_blobs(delimiter='/', prefix='parent/')
-        response = iterator.get_next_page_response()
-        blobs = list(iterator.get_items_from_response(response))
+        iterator._update_page()
+        blobs = list(iterator.page)
         self.assertEqual([blob.name for blob in blobs], ['parent/file11.txt'])
-        self.assertEqual(iterator.page_number, 1)
         self.assertIsNone(iterator.next_page_token)
         self.assertEqual(iterator.prefixes, set(['parent/child/']))
 
@@ -328,11 +325,10 @@ def test_second_level(self):
 
         iterator = self.bucket.list_blobs(delimiter='/',
                                           prefix='parent/child/')
-        response = iterator.get_next_page_response()
-        blobs = list(iterator.get_items_from_response(response))
+        iterator._update_page()
+        blobs = list(iterator.page)
         self.assertEqual([blob.name for blob in blobs],
                          expected_names)
-        self.assertEqual(iterator.page_number, 1)
         self.assertIsNone(iterator.next_page_token)
         self.assertEqual(iterator.prefixes,
                          set(['parent/child/grand/', 'parent/child/other/']))
@@ -345,11 +341,10 @@ def test_third_level(self):
         # Exercise a layer deeper to illustrate this.
         iterator = self.bucket.list_blobs(delimiter='/',
                                           prefix='parent/child/grand/')
-        response = iterator.get_next_page_response()
-        blobs = list(iterator.get_items_from_response(response))
+        iterator._update_page()
+        blobs = list(iterator.page)
         self.assertEqual([blob.name for blob in blobs],
                          ['parent/child/grand/file31.txt'])
-        self.assertEqual(iterator.page_number, 1)
         self.assertIsNone(iterator.next_page_token)
         self.assertEqual(iterator.prefixes, set())