feat(storage): add get notification method

google/cloud/storage/bucket.py

@@ -617,12 +617,13 @@ def blob(
 
     def notification(
         self,
-        topic_name,
+        topic_name=None,
         topic_project=None,
         custom_attributes=None,
         event_types=None,
         blob_name_prefix=None,
         payload_format=NONE_PAYLOAD_FORMAT,
+        notification_id=None,
     ):
         """Factory:  create a notification resource for the bucket.
 
@@ -632,12 +633,13 @@ def notification(
         """
         return BucketNotification(
             self,
-            topic_name,
+            topic_name=topic_name,
             topic_project=topic_project,
             custom_attributes=custom_attributes,
             event_types=event_types,
             blob_name_prefix=blob_name_prefix,
             payload_format=payload_format,
+            notification_id=notification_id,
         )
 
     def exists(self, client=None, timeout=_DEFAULT_TIMEOUT):
@@ -1021,6 +1023,44 @@ def list_notifications(self, client=None, timeout=_DEFAULT_TIMEOUT):
         iterator.bucket = self
         return iterator
 
+    def get_notification(self, notification_id, client=None, timeout=_DEFAULT_TIMEOUT):
+        """Get Pub / Sub notification for this bucket.
+
+        See:
+        https://cloud.google.com/storage/docs/json_api/v1/notifications/get
+
+        If :attr:`user_project` is set, bills the API request to that project.
+
+        :type notification_id: str
+        :param notification_id: The notification id to retrieve the notification configuration.
+
+        :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.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
+
+        :rtype: :class:`.BucketNotification`
+        :returns: notification instance.
+
+        Example:
+            Get notification using notification id.
+
+            >>> from google.cloud import storage
+            >>> client = storage.Client()
+            >>> bucket = client.get_bucket('my-bucket-name')  # API request.
+            >>> notification = bucket.get_notification(notification_id='id')  # API request.
+
+        """
+        notification = self.notification(notification_id=notification_id)
+        notification.reload(client=client, timeout=timeout)
+        return notification
+
     def delete(self, force=False, client=None, timeout=_DEFAULT_TIMEOUT):
         """Delete this bucket.
 

google/cloud/storage/notification.py

@@ -50,7 +50,8 @@ class BucketNotification(object):
     :param bucket: Bucket to which the notification is bound.
 
     :type topic_name: str
-    :param topic_name: Topic name to which notifications are published.
+    :param topic_name:
+        (Optional) Topic name to which notifications are published.
 
     :type topic_project: str
     :param topic_project:
@@ -63,7 +64,7 @@ class BucketNotification(object):
 
     :type event_types: list(str)
     :param event_types:
-        (Optional) event types for which notificatin events are published.
+        (Optional) event types for which notification events are published.
 
     :type blob_name_prefix: str
     :param blob_name_prefix:
@@ -73,17 +74,22 @@ class BucketNotification(object):
     :type payload_format: str
     :param payload_format:
         (Optional) format of payload for notification events.
+
+    :type notification_id: str
+    :param notification_id:
+        (Optional) The ID of the notification.
     """
 
     def __init__(
         self,
         bucket,
-        topic_name,
+        topic_name=None,
         topic_project=None,
         custom_attributes=None,
         event_types=None,
         blob_name_prefix=None,
         payload_format=NONE_PAYLOAD_FORMAT,
+        notification_id=None,
     ):
         self._bucket = bucket
         self._topic_name = topic_name
@@ -107,6 +113,9 @@ def __init__(
         if blob_name_prefix is not None:
             self._properties["object_name_prefix"] = blob_name_prefix
 
+        if notification_id is not None:
+            self._properties["id"] = notification_id
+
         self._properties["payload_format"] = payload_format
 
     @classmethod

tests/system/test_system.py

@@ -1433,6 +1433,7 @@ def test_notification_minimal(self):
         bucket = retry_429_503(Config.CLIENT.create_bucket)(new_bucket_name)
         self.case_buckets_to_delete.append(new_bucket_name)
         self.assertEqual(list(bucket.list_notifications()), [])
+
         notification = bucket.notification(self.TOPIC_NAME)
         retry_429_503(notification.create)()
         try:
@@ -1449,7 +1450,7 @@ def test_notification_explicit(self):
         bucket = retry_429_503(Config.CLIENT.create_bucket)(new_bucket_name)
         self.case_buckets_to_delete.append(new_bucket_name)
         notification = bucket.notification(
-            self.TOPIC_NAME,
+            topic_name=self.TOPIC_NAME,
             custom_attributes=self.CUSTOM_ATTRIBUTES,
             event_types=self.event_types(),
             blob_name_prefix=self.BLOB_NAME_PREFIX,
@@ -1463,6 +1464,7 @@ def test_notification_explicit(self):
             self.assertEqual(notification.event_types, self.event_types())
             self.assertEqual(notification.blob_name_prefix, self.BLOB_NAME_PREFIX)
             self.assertEqual(notification.payload_format, self.payload_format())
+
         finally:
             notification.delete()
 
@@ -1486,6 +1488,28 @@ def test_notification_w_user_project(self):
         finally:
             notification.delete()
 
+    def test_get_notification(self):
+        new_bucket_name = "get-notification" + unique_resource_id("-")
+        bucket = retry_429_503(Config.CLIENT.create_bucket)(new_bucket_name)
+        self.case_buckets_to_delete.append(new_bucket_name)
+
+        notification = bucket.notification(
+            topic_name=self.TOPIC_NAME,
+            custom_attributes=self.CUSTOM_ATTRIBUTES,
+            payload_format=self.payload_format(),
+        )
+        retry_429_503(notification.create)()
+        try:
+            self.assertTrue(notification.exists())
+            self.assertIsNotNone(notification.notification_id)
+            notification_id = notification.notification_id
+            notification = bucket.get_notification(notification_id)
+            self.assertEqual(notification.notification_id, notification_id)
+            self.assertEqual(notification.custom_attributes, self.CUSTOM_ATTRIBUTES)
+            self.assertEqual(notification.payload_format, self.payload_format())
+        finally:
+            notification.delete()
+
 
 class TestAnonymousClient(unittest.TestCase):
 

tests/unit/test_bucket.py

@@ -826,6 +826,45 @@ def test_list_notifications(self):
                 notification.payload_format, resource.get("payload_format")
             )
 
+    def test_get_notification(self):
+        from google.cloud.storage.notification import _TOPIC_REF_FMT
+        from google.cloud.storage.notification import JSON_API_V1_PAYLOAD_FORMAT
+
+        NAME = "name"
+        ETAG = "FACECABB"
+        NOTIFICATION_ID = "1"
+        SELF_LINK = "https://example.com/notification/1"
+        resources = {
+            "topic": _TOPIC_REF_FMT.format("my-project-123", "topic-1"),
+            "id": NOTIFICATION_ID,
+            "etag": ETAG,
+            "selfLink": SELF_LINK,
+            "payload_format": JSON_API_V1_PAYLOAD_FORMAT,
+        }
+
+        connection = _make_connection(resources)
+        client = _Client(connection, project="my-project-123")
+        bucket = self._make_one(client=client, name=NAME)
+        notification = bucket.get_notification(notification_id=NOTIFICATION_ID)
+
+        self.assertEqual(notification.notification_id, NOTIFICATION_ID)
+        self.assertEqual(notification.etag, ETAG)
+        self.assertEqual(notification.self_link, SELF_LINK)
+        self.assertIsNone(notification.custom_attributes)
+        self.assertIsNone(notification.event_types)
+        self.assertIsNone(notification.blob_name_prefix)
+        self.assertEqual(notification.payload_format, JSON_API_V1_PAYLOAD_FORMAT)
+
+    def test_get_notification_miss(self):
+        from google.cloud.exceptions import NotFound
+
+        response = NotFound("testing")
+        connection = _make_connection(response)
+        client = _Client(connection, project="my-project-123")
+        bucket = self._make_one(client=client, name="name")
+        with self.assertRaises(NotFound):
+            bucket.get_notification(notification_id="1")
+
     def test_delete_miss(self):
         from google.cloud.exceptions import NotFound