Add 'BucketNotification.exists' API wrapper.

storage/google/cloud/storage/notification.py

@@ -14,6 +14,9 @@
 
 """Support for bucket notification resources."""
 
+from google.api.core.exceptions import NotFound
+
+
 OBJECT_FINALIZE_EVENT_TYPE = 'OBJECT_FINALIZE'
 OBJECT_METADATA_UPDATE_EVENT_TYPE = 'OBJECT_METADATA_UPDATE'
 OBJECT_DELETE_EVENT_TYPE = 'OBJECT_DELETE'
@@ -187,6 +190,31 @@ def create(self, client=None):
             data=properties,
         )
 
+    def exists(self, client=None):
+        """Test whether this notification exists.
+
+        See:
+        https://cloud.google.com/storage/docs/json_api/v1/notifications/get
+
+        :type client: :class:`~google.cloud.storage.client.Client` or
+                      ``NoneType``
+        :param client: Optional. The client to use.  If not passed, falls back
+                       to the ``client`` stored on the current bucket.
+
+        :rtype: bool
+        :returns: True, if the notification exists, else False.
+        """
+        if self.notification_id is None:
+            raise ValueError("Notification not intialized by server")
+
+        client = self._require_client(client)
+        try:
+            client._connection.api_request(method='GET', path=self.path)
+        except NotFound:
+            return False
+        else:
+            return True
+
     def delete(self, client=None):
         """Delete this notification.
 

storage/tests/unit/test_notification.py

@@ -244,6 +244,54 @@ def test_create_w_explicit_client(self):
             data=data,
         )
 
+    def test_exists_wo_notification_id(self):
+        client = self._make_client()
+        bucket = self._make_bucket(client)
+        notification = self._make_one(
+            bucket, self.TOPIC_NAME)
+
+        with self.assertRaises(ValueError):
+            notification.exists()
+
+    def test_exists_miss(self):
+        from google.cloud.exceptions import NotFound
+
+        NOTIFICATION_ID = '123'
+        client = self._make_client()
+        bucket = self._make_bucket(client)
+        notification = self._make_one(bucket, self.TOPIC_NAME)
+        notification._properties['id'] = NOTIFICATION_ID
+        api_request = client._connection.api_request
+        api_request.side_effect = NotFound('testing')
+
+        self.assertFalse(notification.exists())
+
+        path = '/b/{}/notificationConfigs/{}'.format(
+            self.BUCKET_NAME, NOTIFICATION_ID)
+        api_request.assert_called_once_with(
+            method='GET',
+            path=path,
+        )
+
+    def test_exists_hit(self):
+        NOTIFICATION_ID = '123'
+        client = self._make_client()
+        bucket = self._make_bucket(client)
+        alt_client = self._make_client()
+        notification = self._make_one(bucket, self.TOPIC_NAME)
+        notification._properties['id'] = NOTIFICATION_ID
+        api_request = client._connection.api_request
+        api_request.return_value = None
+
+        self.assertTrue(notification.exists(client=client))
+
+        path = '/b/{}/notificationConfigs/{}'.format(
+            self.BUCKET_NAME, NOTIFICATION_ID)
+        api_request.assert_called_once_with(
+            method='GET',
+            path=path,
+        )
+
     def test_delete_wo_notification_id(self):
         client = self._make_client()
         bucket = self._make_bucket(client)