diff --git a/gcloud/logging/client.py b/gcloud/logging/client.py
index 4f87fdf6cf71..1713d8d47538 100644
--- a/gcloud/logging/client.py
+++ b/gcloud/logging/client.py
@@ -154,3 +154,39 @@ def sink(self, name, filter_, destination):
         :returns: Sink created with the current client.
         """
         return Sink(name, filter_, destination, client=self)
+
+    def list_sinks(self, page_size=None, page_token=None):
+        """List sinks for the project associated with this client.
+
+        See:
+        https://cloud.google.com/logging/docs/api/ref_v2beta1/rest/v2beta1/projects.sinks/list
+
+        :type page_size: int
+        :param page_size: maximum number of sinks to return, If not passed,
+                          defaults to a value set by the API.
+
+        :type page_token: string
+        :param page_token: opaque marker for the next "page" of sinks. If not
+                           passed, the API will return the first page of
+                           sinks.
+
+        :rtype: tuple, (list, str)
+        :returns: list of :class:`gcloud.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``).
+        """
+        params = {}
+
+        if page_size is not None:
+            params['pageSize'] = page_size
+
+        if page_token is not None:
+            params['pageToken'] = page_token
+
+        path = '/projects/%s/sinks' % (self.project,)
+        resp = self.connection.api_request(method='GET', path=path,
+                                           query_params=params)
+        sinks = [Sink.from_api_repr(resource, self)
+                 for resource in resp.get('sinks', ())]
+        return sinks, resp.get('nextPageToken')
diff --git a/gcloud/logging/sink.py b/gcloud/logging/sink.py
index 4a2d19699bc4..983861e61d3a 100644
--- a/gcloud/logging/sink.py
+++ b/gcloud/logging/sink.py
@@ -14,9 +14,36 @@
 
 """Define Logging API Sinks."""
 
+import re
+
+from gcloud._helpers import _name_from_project_path
 from gcloud.exceptions import NotFound
 
 
+_SINK_TEMPLATE = re.compile(r"""
+    projects/            # static prefix
+    (?P<project>[^/]+)   # initial letter, wordchars + hyphen
+    /sinks/              # static midfix
+    (?P<name>[^/]+)      # initial letter, wordchars + allowed punc
+""", re.VERBOSE)
+
+
+def _sink_name_from_path(path, project):
+    """Validate a sink URI path and get the sink name.
+    :type path: string
+    :param path: URI path for a sink API request.
+    :type project: string
+    :param project: The project associated with the request. It is
+                    included for validation purposes.
+    :rtype: string
+    :returns: Metric name parsed from ``path``.
+    :raises: :class:`ValueError` if the ``path`` is ill-formed or if
+             the project from the ``path`` does not agree with the
+             ``project`` passed in.
+    """
+    return _name_from_project_path(path, project, _SINK_TEMPLATE)
+
+
 class Sink(object):
     """Sinks represent filtered exports for log entries.
 
@@ -63,11 +90,35 @@ def path(self):
         """URL path for the sink's APIs"""
         return '/%s' % (self.full_name)
 
+    @classmethod
+    def from_api_repr(cls, resource, client):
+        """Factory:  construct a sink given its API representation
+
+        :type resource: dict
+        :param resource: sink resource representation returned from the API
+
+        :type client: :class:`gcloud.pubsub.client.Client`
+        :param client: Client which holds credentials and project
+                       configuration for the sink.
+
+        :rtype: :class:`gcloud.logging.sink.Sink`
+        :returns: Sink parsed from ``resource``.
+        :raises: :class:`ValueError` if ``client`` is not ``None`` and the
+                 project from the resource does not agree with the project
+                 from the client.
+        """
+        sink_name = _sink_name_from_path(resource['name'], client.project)
+        filter_ = resource['filter']
+        destination = resource['destination']
+        return cls(sink_name, filter_, destination, client=client)
+
     def _require_client(self, client):
         """Check client or verify over-ride.
+
         :type client: :class:`gcloud.logging.client.Client` or ``NoneType``
         :param client: the client to use.  If not passed, falls back to the
                        ``client`` stored on the current sink.
+
         :rtype: :class:`gcloud.logging.client.Client`
         :returns: The client passed in or the currently bound client.
         """
diff --git a/gcloud/logging/test_client.py b/gcloud/logging/test_client.py
index a56640cc9e1c..94a9997e978e 100644
--- a/gcloud/logging/test_client.py
+++ b/gcloud/logging/test_client.py
@@ -171,6 +171,105 @@ def test_sink(self):
         self.assertTrue(sink.client is client)
         self.assertEqual(sink.project, self.PROJECT)
 
+    def test_list_sinks_no_paging(self):
+        from gcloud.logging.sink import Sink
+        PROJECT = 'PROJECT'
+        CREDS = _Credentials()
+
+        CLIENT_OBJ = self._makeOne(project=PROJECT, credentials=CREDS)
+
+        SINK_NAME = 'sink_name'
+        FILTER = 'logName:syslog AND severity>=ERROR'
+        SINK_PATH = 'projects/%s/sinks/%s' % (PROJECT, SINK_NAME)
+
+        RETURNED = {
+            'sinks': [{
+                'name': SINK_PATH,
+                'filter': FILTER,
+                'destination': self.DESTINATION_URI,
+            }],
+        }
+        # Replace the connection on the client with one of our own.
+        CLIENT_OBJ.connection = _Connection(RETURNED)
+
+        # Execute request.
+        sinks, next_page_token = CLIENT_OBJ.list_sinks()
+        # Test values are correct.
+        self.assertEqual(len(sinks), 1)
+        sink = sinks[0]
+        self.assertTrue(isinstance(sink, Sink))
+        self.assertEqual(sink.name, SINK_NAME)
+        self.assertEqual(sink.filter_, FILTER)
+        self.assertEqual(sink.destination, self.DESTINATION_URI)
+        self.assertEqual(next_page_token, None)
+        self.assertEqual(len(CLIENT_OBJ.connection._requested), 1)
+        req = CLIENT_OBJ.connection._requested[0]
+        self.assertEqual(req['method'], 'GET')
+        self.assertEqual(req['path'], '/projects/%s/sinks' % (PROJECT,))
+        self.assertEqual(req['query_params'], {})
+
+    def test_list_sinks_with_paging(self):
+        from gcloud.logging.sink import Sink
+        PROJECT = 'PROJECT'
+        CREDS = _Credentials()
+
+        CLIENT_OBJ = self._makeOne(project=PROJECT, credentials=CREDS)
+
+        SINK_NAME = 'sink_name'
+        FILTER = 'logName:syslog AND severity>=ERROR'
+        SINK_PATH = 'projects/%s/sinks/%s' % (PROJECT, SINK_NAME)
+        TOKEN1 = 'TOKEN1'
+        TOKEN2 = 'TOKEN2'
+        SIZE = 1
+        RETURNED = {
+            'sinks': [{
+                'name': SINK_PATH,
+                'filter': FILTER,
+                'destination': self.DESTINATION_URI,
+            }],
+            'nextPageToken': TOKEN2,
+        }
+        # Replace the connection on the client with one of our own.
+        CLIENT_OBJ.connection = _Connection(RETURNED)
+
+        # Execute request.
+        sinks, next_page_token = CLIENT_OBJ.list_sinks(SIZE, TOKEN1)
+        # Test values are correct.
+        self.assertEqual(len(sinks), 1)
+        sink = sinks[0]
+        self.assertTrue(isinstance(sink, Sink))
+        self.assertEqual(sink.name, SINK_NAME)
+        self.assertEqual(sink.filter_, FILTER)
+        self.assertEqual(sink.destination, self.DESTINATION_URI)
+        self.assertEqual(next_page_token, TOKEN2)
+        self.assertEqual(len(CLIENT_OBJ.connection._requested), 1)
+        req = CLIENT_OBJ.connection._requested[0]
+        self.assertEqual(req['method'], 'GET')
+        self.assertEqual(req['path'], '/projects/%s/sinks' % (PROJECT,))
+        self.assertEqual(req['query_params'],
+                         {'pageSize': SIZE, 'pageToken': TOKEN1})
+
+    def test_list_sinks_missing_key(self):
+        PROJECT = 'PROJECT'
+        CREDS = _Credentials()
+
+        CLIENT_OBJ = self._makeOne(project=PROJECT, credentials=CREDS)
+
+        RETURNED = {}
+        # Replace the connection on the client with one of our own.
+        CLIENT_OBJ.connection = _Connection(RETURNED)
+
+        # Execute request.
+        sinks, next_page_token = CLIENT_OBJ.list_sinks()
+        # Test values are correct.
+        self.assertEqual(len(sinks), 0)
+        self.assertEqual(next_page_token, None)
+        self.assertEqual(len(CLIENT_OBJ.connection._requested), 1)
+        req = CLIENT_OBJ.connection._requested[0]
+        self.assertEqual(req['method'], 'GET')
+        self.assertEqual(req['path'], '/projects/%s/sinks' % PROJECT)
+        self.assertEqual(req['query_params'], {})
+
 
 class _Credentials(object):
 
diff --git a/gcloud/logging/test_sink.py b/gcloud/logging/test_sink.py
index 69c3a4e6adaf..e78860a772a3 100644
--- a/gcloud/logging/test_sink.py
+++ b/gcloud/logging/test_sink.py
@@ -15,6 +15,38 @@
 import unittest2
 
 
+class Test__sink_name_from_path(unittest2.TestCase):
+
+    def _callFUT(self, path, project):
+        from gcloud.logging.sink import _sink_name_from_path
+        return _sink_name_from_path(path, project)
+
+    def test_invalid_path_length(self):
+        PATH = 'projects/foo'
+        PROJECT = None
+        self.assertRaises(ValueError, self._callFUT, PATH, PROJECT)
+
+    def test_invalid_path_format(self):
+        SINK_NAME = 'SINK_NAME'
+        PROJECT = 'PROJECT'
+        PATH = 'foo/%s/bar/%s' % (PROJECT, SINK_NAME)
+        self.assertRaises(ValueError, self._callFUT, PATH, PROJECT)
+
+    def test_invalid_project(self):
+        SINK_NAME = 'SINK_NAME'
+        PROJECT1 = 'PROJECT1'
+        PROJECT2 = 'PROJECT2'
+        PATH = 'projects/%s/sinks/%s' % (PROJECT1, SINK_NAME)
+        self.assertRaises(ValueError, self._callFUT, PATH, PROJECT2)
+
+    def test_valid_data(self):
+        SINK_NAME = 'SINK_NAME'
+        PROJECT = 'PROJECT'
+        PATH = 'projects/%s/sinks/%s' % (PROJECT, SINK_NAME)
+        sink_name = self._callFUT(PATH, PROJECT)
+        self.assertEqual(sink_name, SINK_NAME)
+
+
 class TestSink(unittest2.TestCase):
 
     PROJECT = 'test-project'
@@ -43,6 +75,54 @@ def test_ctor(self):
         self.assertEqual(sink.full_name, FULL)
         self.assertEqual(sink.path, '/%s' % (FULL,))
 
+    def test_from_api_repr_minimal(self):
+        CLIENT = _Client(project=self.PROJECT)
+        FULL = 'projects/%s/sinks/%s' % (self.PROJECT, self.SINK_NAME)
+        RESOURCE = {
+            'name': FULL,
+            'filter': self.FILTER,
+            'destination': self.DESTINATION_URI,
+        }
+        klass = self._getTargetClass()
+        sink = klass.from_api_repr(RESOURCE, client=CLIENT)
+        self.assertEqual(sink.name, self.SINK_NAME)
+        self.assertEqual(sink.filter_, self.FILTER)
+        self.assertEqual(sink.destination, self.DESTINATION_URI)
+        self.assertTrue(sink._client is CLIENT)
+        self.assertEqual(sink.project, self.PROJECT)
+        self.assertEqual(sink.full_name, FULL)
+
+    def test_from_api_repr_w_description(self):
+        CLIENT = _Client(project=self.PROJECT)
+        FULL = 'projects/%s/sinks/%s' % (self.PROJECT, self.SINK_NAME)
+        RESOURCE = {
+            'name': FULL,
+            'filter': self.FILTER,
+            'destination': self.DESTINATION_URI,
+        }
+        klass = self._getTargetClass()
+        sink = klass.from_api_repr(RESOURCE, client=CLIENT)
+        self.assertEqual(sink.name, self.SINK_NAME)
+        self.assertEqual(sink.filter_, self.FILTER)
+        self.assertEqual(sink.destination, self.DESTINATION_URI)
+        self.assertTrue(sink._client is CLIENT)
+        self.assertEqual(sink.project, self.PROJECT)
+        self.assertEqual(sink.full_name, FULL)
+
+    def test_from_api_repr_with_mismatched_project(self):
+        PROJECT1 = 'PROJECT1'
+        PROJECT2 = 'PROJECT2'
+        CLIENT = _Client(project=PROJECT1)
+        FULL = 'projects/%s/sinks/%s' % (PROJECT2, self.SINK_NAME)
+        RESOURCE = {
+            'name': FULL,
+            'filter': self.FILTER,
+            'destination': self.DESTINATION_URI,
+        }
+        klass = self._getTargetClass()
+        self.assertRaises(ValueError, klass.from_api_repr,
+                          RESOURCE, client=CLIENT)
+
     def test_create_w_bound_client(self):
         FULL = 'projects/%s/sinks/%s' % (self.PROJECT, self.SINK_NAME)
         RESOURCE = {