Add 'BucketNotification' class.

storage/google/cloud/storage/notification.py

@@ -0,0 +1,135 @@
+# Copyright 2017 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Support for bucket notification resources."""
+
+OBJECT_FINALIZE_EVENT_TYPE = 'OBJECT_FINALIZE'
+OBJECT_METADATA_UPDATE_EVENT_TYPE = 'OBJECT_METADATA_UPDATE'
+OBJECT_DELETE_EVENT_TYPE = 'OBJECT_DELETE'
+OBJECT_ARCHIVE_EVENT_TYPE = 'OBJECT_ARCHIVE'
+
+JSON_API_V1_PAYLOAD_FORMAT = 'JSON_API_V1'
+NONE_PAYLOAD_FORMAT = 'NONE'
+
+
+class BucketNotification(object):
+    """Represent a single notification resource for a bucket.
+
+    See: https://cloud.google.com/storage/docs/json_api/v1/notifications
+
+    :type bucket: :class:`google.cloud.storage.bucket.Bucket`
+    :param bucket: Bucket to which the notification is bound.
+
+    :type topic_name: str
+    :param topic_name: Topic name to which notifications are published.
+
+    :type topic_project: str
+    :param topic_project:
+        (Optional) project ID of topic to which notifications are published.
+        If not passed, uses the project ID of the bucket's client.
+
+    :type custom_attributes: dict
+    :param custom_attributes:
+        (Optional) additional attributes passed with notification events.
+
+    :type event_types: list(str)
+    :param event_types:
+        (Optional) event types for which notificatin events are published.
+
+    :type blob_name_prefix: str
+    :param blob_name_prefix:
+        (Optional) prefix of blob names for which notification events are
+        published..
+
+    :type payload_format: str
+    :param payload_format:
+        (Optional) format of payload for notification events.
+    """
+    def __init__(self, bucket, topic_name,
+                 topic_project=None, custom_attributes=None, event_types=None,
+                 blob_name_prefix=None, payload_format=None):
+        self._bucket = bucket
+        self._topic_name = topic_name
+
+        if topic_project is None:
+            topic_project = bucket.client.project
+        self._topic_project = topic_project
+
+        self._properties = {}
+
+        if custom_attributes is not None:
+            self._properties['custom_attributes'] = custom_attributes
+
+        if event_types is not None:
+            self._properties['event_types'] = event_types
+
+        if blob_name_prefix is not None:
+            self._properties['blob_name_prefix'] = blob_name_prefix
+
+        if payload_format is not None:
+            self._properties['payload_format'] = payload_format
+
+    @property
+    def bucket(self):
+        """Bucket to which the notification is bound."""
+        return self._bucket
+
+    @property
+    def topic_name(self):
+        """Topic name to which notifications are published."""
+        return self._topic_name
+
+    @property
+    def topic_project(self):
+        """Project ID of topic to which notifications are published.
+        """
+        return self._topic_project
+
+    @property
+    def custom_attributes(self):
+        """Custom attributes passed with notification events.
+        """
+        return self._properties.get('custom_attributes')
+
+    @property
+    def event_types(self):
+        """Event types for which notification events are published.
+        """
+        return self._properties.get('event_types')
+
+    @property
+    def blob_name_prefix(self):
+        """Prefix of blob names for which notification events are published.
+        """
+        return self._properties.get('blob_name_prefix')
+
+    @property
+    def payload_format(self):
+        """Format of payload of notification events."""
+        return self._properties.get('payload_format')
+
+    @property
+    def notification_id(self):
+        """Server-set ID of notification resource."""
+        return self._properties.get('id')
+
+    @property
+    def etag(self):
+        """Server-set ETag of notification resource."""
+        return self._properties.get('etag')
+
+    @property
+    def self_link(self):
+        """Server-set ETag of notification resource."""
+        return self._properties.get('selfLink')

storage/tests/unit/test_notification.py

@@ -0,0 +1,131 @@
+# Copyright 2017 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+
+import mock
+
+
+class TestBucketNotification(unittest.TestCase):
+
+    BUCKET_NAME = 'test-bucket'
+    BUCKET_PROJECT = 'bucket-project-123'
+    TOPIC_NAME = 'test-topic'
+    TOPIC_ALT_PROJECT = 'topic-project-456'
+
+    @staticmethod
+    def _get_target_class():
+        from google.cloud.storage.notification import BucketNotification
+
+        return BucketNotification
+
+    def _make_one(self, *args, **kw):
+        return self._get_target_class()(*args, **kw)
+
+    def _make_client(self, project=BUCKET_PROJECT):
+        from google.cloud.storage.client import Client
+
+        return mock.Mock(project=project, spec=Client)
+
+    def _make_bucket(self, client, name=BUCKET_NAME):
+        from google.cloud.storage.bucket import Bucket
+
+        return mock.Mock(client=client, name=name, spec=Bucket)
+
+    def test_ctor_defaults(self):
+        client = self._make_client()
+        bucket = self._make_bucket(client)
+
+        notification = self._make_one(
+            bucket, self.TOPIC_NAME)
+
+        self.assertIs(notification.bucket, bucket)
+        self.assertEqual(notification.topic_name, self.TOPIC_NAME)
+        self.assertEqual(notification.topic_project, self.BUCKET_PROJECT)
+        self.assertIsNone(notification.custom_attributes)
+        self.assertIsNone(notification.event_types)
+        self.assertIsNone(notification.blob_name_prefix)
+        self.assertIsNone(notification.payload_format)
+
+    def test_ctor_explicit(self):
+        from google.cloud.storage.notification import (
+            OBJECT_FINALIZE_EVENT_TYPE,
+            OBJECT_DELETE_EVENT_TYPE,
+            JSON_API_V1_PAYLOAD_FORMAT)
+
+        client = self._make_client()
+        bucket = self._make_bucket(client)
+        CUSTOM_ATTRIBUTES = {
+            'attr1': 'value1',
+            'attr2': 'value2',
+        }
+        EVENT_TYPES = [OBJECT_FINALIZE_EVENT_TYPE, OBJECT_DELETE_EVENT_TYPE]
+        BLOB_NAME_PREFIX = 'blob-name-prefix/'
+
+        notification = self._make_one(
+            bucket, self.TOPIC_NAME,
+            topic_project=self.TOPIC_ALT_PROJECT,
+            custom_attributes=CUSTOM_ATTRIBUTES,
+            event_types=EVENT_TYPES,
+            blob_name_prefix=BLOB_NAME_PREFIX,
+            payload_format=JSON_API_V1_PAYLOAD_FORMAT,
+        )
+
+        self.assertIs(notification.bucket, bucket)
+        self.assertEqual(notification.topic_name, self.TOPIC_NAME)
+        self.assertEqual(notification.topic_project, self.TOPIC_ALT_PROJECT)
+        self.assertEqual(notification.custom_attributes, CUSTOM_ATTRIBUTES)
+        self.assertEqual(notification.event_types, EVENT_TYPES)
+        self.assertEqual(notification.blob_name_prefix, BLOB_NAME_PREFIX)
+        self.assertEqual(
+            notification.payload_format, JSON_API_V1_PAYLOAD_FORMAT)
+
+    def test_notification_id(self):
+        client = self._make_client()
+        bucket = self._make_bucket(client)
+        NOTIFICATION_ID = '123'
+
+        notification = self._make_one(
+            bucket, self.TOPIC_NAME)
+
+        self.assertIsNone(notification.notification_id)
+
+        notification._properties['id'] = NOTIFICATION_ID
+        self.assertEqual(notification.notification_id, NOTIFICATION_ID)
+
+    def test_etag(self):
+        client = self._make_client()
+        bucket = self._make_bucket(client)
+        ETAG = 'DEADBEEF'
+
+        notification = self._make_one(
+            bucket, self.TOPIC_NAME)
+
+        self.assertIsNone(notification.etag)
+
+        notification._properties['etag'] = ETAG
+        self.assertEqual(notification.etag, ETAG)
+
+    def test_self_link(self):
+        client = self._make_client()
+        bucket = self._make_bucket(client)
+        SELF_LINK = 'https://example.com/notification/123'
+
+        notification = self._make_one(
+            bucket, self.TOPIC_NAME)
+
+        self.assertIsNone(notification.self_link)
+
+        notification._properties['selfLink'] = SELF_LINK
+        self.assertEqual(notification.self_link, SELF_LINK)