Merge pull request #2652 from dhermes/logging-iterators-sinks

Updating list_sinks() to Iterator pattern.

docs/logging-usage.rst

@@ -342,8 +342,7 @@ List all sinks for a project:
 
    >>> from google.cloud import logging
    >>> client = logging.Client()
-   >>> sinks, token = client.list_sinks()
-   >>> for sink in sinks:
+   >>> for sink in client.list_sinks():  # API call(s)
    ...     print('%s: %s' % (sink.name, sink.destination))
    robots-storage: storage.googleapis.com/my-bucket-name
    robots-bq: bigquery.googleapis.com/projects/my-project/datasets/my-dataset

logging/google/cloud/logging/_gax.py

@@ -32,6 +32,7 @@
 from google.cloud.exceptions import NotFound
 from google.cloud.iterator import GAXIterator
 from google.cloud.logging._helpers import entry_from_resource
+from google.cloud.logging.sink import Sink
 
 
 class _LoggingAPI(object):
@@ -178,10 +179,7 @@ def list_sinks(self, project, page_size=0, page_token=None):
         path = 'projects/%s' % (project,)
         page_iter = self._gax_api.list_sinks(path, page_size=page_size,
                                              options=options)
-        sinks = [MessageToDict(log_sink_pb)
-                 for log_sink_pb in page_iter.next()]
-        token = page_iter.page_token or None
-        return sinks, token
+        return GAXIterator(self._client, page_iter, _item_to_sink)
 
     def sink_create(self, project, sink_name, filter_, destination):
         """API call:  create a sink resource.
@@ -481,3 +479,20 @@ def _item_to_entry(iterator, entry_pb, loggers):
     """
     resource = MessageToDict(entry_pb)
     return entry_from_resource(resource, iterator.client, loggers)
+
+
+def _item_to_sink(iterator, log_sink_pb):
+    """Convert a sink protobuf to the native object.
+
+    :type iterator: :class:`~google.cloud.iterator.Iterator`
+    :param iterator: The iterator that is currently in use.
+
+    :type log_sink_pb:
+        :class:`~google.logging.v2.logging_config_pb2.LogSink`
+    :param log_sink_pb: Sink protobuf returned from the API.
+
+    :rtype: :class:`~google.cloud.logging.sink.Sink`
+    :returns: The next sink in the page.
+    """
+    resource = MessageToDict(log_sink_pb)
+    return Sink.from_api_repr(resource, iterator.client)

logging/google/cloud/logging/_helpers.py

@@ -24,7 +24,7 @@ def entry_from_resource(resource, client, loggers):
     """Detect correct entry type from resource and instantiate.
 
     :type resource: dict
-    :param resource: one entry resource from API response
+    :param resource: One entry resource from API response.
 
     :type client: :class:`~google.cloud.logging.client.Client`
     :param client: Client that owns the log entry.
@@ -45,4 +45,4 @@ def entry_from_resource(resource, client, loggers):
     elif 'protoPayload' in resource:
         return ProtobufEntry.from_api_repr(resource, client, loggers)
 
-    raise ValueError('Cannot parse log entry resource')
+    raise ValueError('Cannot parse log entry resource.')

logging/google/cloud/logging/client.py

@@ -222,17 +222,13 @@ def list_sinks(self, page_size=None, page_token=None):
                            passed, the API will return the first page of
                            sinks.
 
-        :rtype: tuple, (list, str)
-        :returns: list of :class:`google.cloud.logging.sink.Sink`, plus a
-                  "next page token" string:  if not None, indicates that
-                  more sinks can be retrieved with another call (pass that
-                  value as ``page_token``).
+        :rtype: :class:`~google.cloud.iterator.Iterator`
+        :returns: Iterator of
+                  :class:`~google.cloud.logging.sink.Sink`
+                  accessible to the current client.
         """
-        resources, token = self.sinks_api.list_sinks(
+        return self.sinks_api.list_sinks(
             self.project, page_size, page_token)
-        sinks = [Sink.from_api_repr(resource, self)
-                 for resource in resources]
-        return sinks, token
 
     def metric(self, name, filter_=None, description=''):
         """Creates a metric bound to the current client.

logging/google/cloud/logging/connection.py

@@ -19,6 +19,7 @@
 from google.cloud import connection as base_connection
 from google.cloud.iterator import HTTPIterator
 from google.cloud.logging._helpers import entry_from_resource
+from google.cloud.logging.sink import Sink
 
 
 class Connection(base_connection.JSONConnection):
@@ -209,24 +210,21 @@ def list_sinks(self, project, page_size=None, page_token=None):
                            passed, the API will return the first page of
                            sinks.
 
-        :rtype: tuple, (list, str)
-        :returns: list of mappings, plus a "next page token" string:
-                  if not None, indicates that more sinks can be retrieved
-                  with another call (pass that value as ``page_token``).
+        :rtype: :class:`~google.cloud.iterator.Iterator`
+        :returns: Iterator of
+                  :class:`~google.cloud.logging.sink.Sink`
+                  accessible to the current API.
         """
-        params = {}
+        extra_params = {}
 
         if page_size is not None:
-            params['pageSize'] = page_size
-
-        if page_token is not None:
-            params['pageToken'] = page_token
+            extra_params['pageSize'] = page_size
 
         path = '/projects/%s/sinks' % (project,)
-        resp = self._connection.api_request(
-            method='GET', path=path, query_params=params)
-        sinks = resp.get('sinks', ())
-        return sinks, resp.get('nextPageToken')
+        return HTTPIterator(
+            client=self._client, path=path,
+            item_to_value=_item_to_sink, items_key='sinks',
+            page_token=page_token, extra_params=extra_params)
 
     def sink_create(self, project, sink_name, filter_, destination):
         """API call:  create a sink resource.
@@ -484,3 +482,18 @@ def _item_to_entry(iterator, resource, loggers):
     :returns: The next log entry in the page.
     """
     return entry_from_resource(resource, iterator.client, loggers)
+
+
+def _item_to_sink(iterator, resource):
+    """Convert a sink resource to the native object.
+
+    :type iterator: :class:`~google.cloud.iterator.Iterator`
+    :param iterator: The iterator that is currently in use.
+
+    :type resource: dict
+    :param resource: Sink JSON resource returned from the API.
+
+    :rtype: :class:`~google.cloud.logging.sink.Sink`
+    :returns: The next sink in the page.
+    """
+    return Sink.from_api_repr(resource, iterator.client)

logging/unit_tests/test__gax.py

@@ -612,56 +612,72 @@ def test_ctor(self):
         self.assertIs(api._client, client)
 
     def test_list_sinks_no_paging(self):
+        import six
         from google.gax import INITIAL_PAGE
-        from google.cloud._testing import _GAXPageIterator
         from google.logging.v2.logging_config_pb2 import LogSink
+        from google.cloud._testing import _GAXPageIterator
+        from google.cloud.logging.sink import Sink
 
         TOKEN = 'TOKEN'
-        SINKS = [{
-            'name': self.SINK_PATH,
-            'filter': self.FILTER,
-            'destination': self.DESTINATION_URI,
-        }]
         sink_pb = LogSink(name=self.SINK_PATH,
                           destination=self.DESTINATION_URI,
                           filter=self.FILTER)
         response = _GAXPageIterator([sink_pb], page_token=TOKEN)
         gax_api = _GAXSinksAPI(_list_sinks_response=response)
-        api = self._makeOne(gax_api, None)
+        client = object()
+        api = self._makeOne(gax_api, client)
 
-        sinks, token = api.list_sinks(self.PROJECT)
+        iterator = api.list_sinks(self.PROJECT)
+        page = six.next(iterator.pages)
+        sinks = list(page)
+        token = iterator.next_page_token
 
-        self.assertEqual(sinks, SINKS)
+        # First check the token.
         self.assertEqual(token, TOKEN)
+        # Then check the sinks returned.
+        self.assertEqual(len(sinks), 1)
+        sink = sinks[0]
+        self.assertIsInstance(sink, Sink)
+        self.assertEqual(sink.name, self.SINK_PATH)
+        self.assertEqual(sink.filter_, self.FILTER)
+        self.assertEqual(sink.destination, self.DESTINATION_URI)
+        self.assertIs(sink.client, client)
 
         project, page_size, options = gax_api._list_sinks_called_with
         self.assertEqual(project, self.PROJECT_PATH)
         self.assertEqual(page_size, 0)
         self.assertEqual(options.page_token, INITIAL_PAGE)
 
     def test_list_sinks_w_paging(self):
-        from google.cloud._testing import _GAXPageIterator
         from google.logging.v2.logging_config_pb2 import LogSink
+        from google.cloud._testing import _GAXPageIterator
+        from google.cloud.logging.sink import Sink
 
         TOKEN = 'TOKEN'
         PAGE_SIZE = 42
-        SINKS = [{
-            'name': self.SINK_PATH,
-            'filter': self.FILTER,
-            'destination': self.DESTINATION_URI,
-        }]
         sink_pb = LogSink(name=self.SINK_PATH,
                           destination=self.DESTINATION_URI,
                           filter=self.FILTER)
         response = _GAXPageIterator([sink_pb])
         gax_api = _GAXSinksAPI(_list_sinks_response=response)
-        api = self._makeOne(gax_api, None)
+        client = object()
+        api = self._makeOne(gax_api, client)
 
-        sinks, token = api.list_sinks(
+        iterator = api.list_sinks(
             self.PROJECT, page_size=PAGE_SIZE, page_token=TOKEN)
+        sinks = list(iterator)
+        token = iterator.next_page_token
 
-        self.assertEqual(sinks, SINKS)
+        # First check the token.
         self.assertIsNone(token)
+        # Then check the sinks returned.
+        self.assertEqual(len(sinks), 1)
+        sink = sinks[0]
+        self.assertIsInstance(sink, Sink)
+        self.assertEqual(sink.name, self.SINK_PATH)
+        self.assertEqual(sink.filter_, self.FILTER)
+        self.assertEqual(sink.destination, self.DESTINATION_URI)
+        self.assertIs(sink.client, client)
 
         project, page_size, options = gax_api._list_sinks_called_with
         self.assertEqual(project, self.PROJECT_PATH)

logging/unit_tests/test_client.py

@@ -365,7 +365,9 @@ def test_sink_explicit(self):
         self.assertEqual(sink.project, self.PROJECT)
 
     def test_list_sinks_no_paging(self):
+        import six
         from google.cloud.logging.sink import Sink
+
         PROJECT = 'PROJECT'
         TOKEN = 'TOKEN'
         SINK_NAME = 'sink_name'
@@ -375,25 +377,42 @@ def test_list_sinks_no_paging(self):
             'filter': FILTER,
             'destination': self.DESTINATION_URI,
         }]
-        client = self._makeOne(project=PROJECT, credentials=_Credentials())
-        api = client._sinks_api = _DummySinksAPI()
-        api._list_sinks_response = SINKS, TOKEN
+        client = self._makeOne(project=PROJECT, credentials=_Credentials(),
+                               use_gax=False)
+        returned = {
+            'sinks': SINKS,
+            'nextPageToken': TOKEN,
+        }
+        client.connection = _Connection(returned)
 
-        sinks, token = client.list_sinks()
+        iterator = client.list_sinks()
+        page = six.next(iterator.pages)
+        sinks = list(page)
+        token = iterator.next_page_token
 
+        # First check the token.
+        self.assertEqual(token, TOKEN)
+        # Then check the sinks returned.
         self.assertEqual(len(sinks), 1)
         sink = sinks[0]
         self.assertIsInstance(sink, Sink)
         self.assertEqual(sink.name, SINK_NAME)
         self.assertEqual(sink.filter_, FILTER)
         self.assertEqual(sink.destination, self.DESTINATION_URI)
+        self.assertIs(sink.client, client)
 
-        self.assertEqual(token, TOKEN)
-        self.assertEqual(api._list_sinks_called_with,
-                         (PROJECT, None, None))
+        # Verify the mocked transport.
+        called_with = client.connection._called_with
+        path = '/projects/%s/sinks' % (self.PROJECT,)
+        self.assertEqual(called_with, {
+            'method': 'GET',
+            'path': path,
+            'query_params': {},
+        })
 
     def test_list_sinks_with_paging(self):
         from google.cloud.logging.sink import Sink
+
         PROJECT = 'PROJECT'
         SINK_NAME = 'sink_name'
         FILTER = 'logName:syslog AND severity>=ERROR'
@@ -404,21 +423,39 @@ def test_list_sinks_with_paging(self):
             'filter': FILTER,
             'destination': self.DESTINATION_URI,
         }]
-        client = self._makeOne(project=PROJECT, credentials=_Credentials())
-        api = client._sinks_api = _DummySinksAPI()
-        api._list_sinks_response = SINKS, None
+        client = self._makeOne(project=PROJECT, credentials=_Credentials(),
+                               use_gax=False)
+        returned = {
+            'sinks': SINKS,
+        }
+        client.connection = _Connection(returned)
 
-        sinks, token = client.list_sinks(PAGE_SIZE, TOKEN)
+        iterator = client.list_sinks(PAGE_SIZE, TOKEN)
+        sinks = list(iterator)
+        token = iterator.next_page_token
 
+        # First check the token.
+        self.assertIsNone(token)
+        # Then check the sinks returned.
         self.assertEqual(len(sinks), 1)
         sink = sinks[0]
         self.assertIsInstance(sink, Sink)
         self.assertEqual(sink.name, SINK_NAME)
         self.assertEqual(sink.filter_, FILTER)
         self.assertEqual(sink.destination, self.DESTINATION_URI)
-        self.assertIsNone(token)
-        self.assertEqual(api._list_sinks_called_with,
-                         (PROJECT, PAGE_SIZE, TOKEN))
+        self.assertIs(sink.client, client)
+
+        # Verify the mocked transport.
+        called_with = client.connection._called_with
+        path = '/projects/%s/sinks' % (self.PROJECT,)
+        self.assertEqual(called_with, {
+            'method': 'GET',
+            'path': path,
+            'query_params': {
+                'pageSize': PAGE_SIZE,
+                'pageToken': TOKEN,
+            },
+        })
 
     def test_metric_defaults(self):
         from google.cloud.logging.metric import Metric
@@ -513,13 +550,6 @@ def create_scoped(self, scope):
         return self
 
 
-class _DummySinksAPI(object):
-
-    def list_sinks(self, project, page_size, page_token):
-        self._list_sinks_called_with = (project, page_size, page_token)
-        return self._list_sinks_response
-
-
 class _DummyMetricsAPI(object):
 
     def list_metrics(self, project, page_size, page_token):