Merge pull request #742 from tseaver/691-flesh_out_pubsub_topics

Flesh out pubsub topics

Author: tseaver

docs/conf.py

@@ -30,6 +30,7 @@
 extensions = [
     'sphinx.ext.autodoc',
     'sphinx.ext.autosummary',
+    'sphinx.ext.doctest',
     'sphinx.ext.todo',
     'sphinx.ext.viewcode',
 ]

docs/index.rst

@@ -14,6 +14,7 @@
   storage-blobs
   storage-buckets
   storage-acl
+  pubsub-api
 
 
 Getting started

docs/pubsub-api.rst

@@ -0,0 +1,251 @@
+``gcloud.pubsub`` API
+=====================
+
+Connection / Authorization
+--------------------------
+
+- Inferred defaults used to create connection if none configured explicitly:
+
+  - credentials (derived from GAE / GCE environ if present).
+
+  - ``project_id`` (derived from GAE / GCE environ if present).
+
+  - ``scopes``
+
+
+Manage topics for a project
+---------------------------
+
+Create a new topic for the default project:
+
+.. doctest::
+
+   >>> from gcloud.pubsub.topic import Topic
+   >>> topic = Topic('topic_name')
+   >>> topic.create()  # API request
+
+Create a new topic for an explicit project:
+
+.. doctest::
+
+   >>> from gcloud.pubsub.topic import Topic
+   >>> topic = Topic('topic_name', project_id='my.project')
+   >>> topic.create()  # API request
+
+Check for the existance of a topic:
+
+.. doctest::
+
+   >>> from gcloud.pubsub.topic import Topic
+   >>> topic = Topic('topic_name')
+   >>> topic.exists()  # API request
+   True
+
+List topics for the default project:
+
+.. doctest::
+
+   >>> from gcloud import pubsub
+   >>> [topic.name for topic in pubsub.list_topics()]  # API request
+   ['topic_name']
+
+List topics for an explicit project:
+
+.. doctest::
+
+   >>> from gcloud import pubsub
+   >>> topics = pubsub.list_topics(project_id='my.project')  # API request
+   >>> [topic.name for topic in topics]
+   ['topic_name']
+
+Delete a topic:
+
+.. doctest::
+
+   >>> from gcloud.pubsub.topic import Topic
+   >>> topic = Topic('topic_name')
+   >>> topic.delete()  # API request
+
+
+Publish messages to a topic
+---------------------------
+
+Publish a single message to a topic, without attributes:
+
+.. doctest::
+
+   >>> from gcloud.pubsub.topic import Topic
+   >>> topic = Topic('topic_name')
+   >>> topic.publish('this is the message_payload')  # API request
+   <message_id>
+
+Publish a single message to a topic, with attributes:
+
+.. doctest::
+
+   >>> from gcloud.pubsub.topic import Topic
+   >>> topic = Topic('topic_name')
+   >>> topic.publish('this is another message_payload',
+   ...               attr1='value1', attr2='value2')  # API request
+   <message_id>
+
+Publish a set of messages to a topic (as a single request):
+
+.. doctest::
+
+   >>> from gcloud.pubsub.topic import Topic
+   >>> topic = Topic('topic_name')
+   >>> with topic.batch() as batch:
+   ...     batch.publish('this is the first message_payload')
+   ...     batch.publish('this is the second message_payload',
+   ...                   attr1='value1', attr2='value2')
+   >>> list(batch)
+   [<message_id1>, <message_id2>]
+
+.. note::
+
+   The only API request happens during the ``__exit__()`` of the topic
+   used as a context manager.
+
+
+Manage subscriptions to topics
+------------------------------
+
+Create a new pull subscription for a topic:
+
+.. doctest::
+
+   >>> from gcloud.pubsub.topic import Topic
+   >>> from gcloud.pubsub.subscription import Subscription
+   >>> topic = Topic('topic_name')
+   >>> subscription = Subscription('subscription_name', topic)
+   >>> subscription.create()  # API request
+
+Create a new pull subscription for a topic with a non-default ACK deadline:
+
+.. doctest::
+
+   >>> from gcloud.pubsub.topic import Topic
+   >>> from gcloud.pubsub.subscription import Subscription
+   >>> topic = Topic('topic_name')
+   >>> subscription = Subscription('subscription_name', ack_deadline=90)
+   >>> subscription.create()  # API request
+
+Create a new push subscription for a topic:
+
+.. doctest::
+
+   >>> ENDPOINT = 'https://example.com/hook'
+   >>> from gcloud.pubsub.topic import Topic
+   >>> from gcloud.pubsub.subscription import Subscription
+   >>> topic = Topic('topic_name')
+   >>> subscription = Subscription('subscription_name', push_endpoint=ENDPOINT)
+   >>> subscription.create()  # API request
+
+Check for the existence of a subscription:
+
+.. doctest::
+
+   >>> from gcloud.pubsub.topic import Topic
+   >>> from gcloud.pubsub.subscription import Subscription
+   >>> topic = Topic('topic_name')
+   >>> subscription = Subscription('subscription_name', topic)
+   >>> subscription.exists()  # API request
+   True
+
+Convert a pull subscription to push:
+
+.. doctest::
+
+   >>> ENDPOINT = 'https://example.com/hook'
+   >>> from gcloud.pubsub.topic import Topic
+   >>> from gcloud.pubsub.subscription import Subscription
+   >>> topic = Topic('topic_name')
+   >>> subscription = Subscription('subscription_name', topic)
+   >>> subscription.modify_push_configuration(push_endpoint=ENDPOINT)  # API request
+
+Convert a push subscription to pull:
+
+.. doctest::
+
+   >>> ENDPOINT = 'https://example.com/hook'
+   >>> from gcloud.pubsub.topic import Topic
+   >>> topic = Topic('topic_name')
+   >>> subscription = Subscription('subscription_name', topic,
+   ...                             push_endpoint=ENDPOINT)
+   >>> subscription.modify_push_configuration(push_endpoint=None)  # API request
+
+List subscriptions for a topic:
+
+.. doctest::
+
+   >>> from gcloud.pubsub.topic import Topic
+   >>> topic = Topic('topic_name')
+   >>> subscriptions = topic.list_subscriptions()  # API request
+   >>> [subscription.name for subscription in subscriptions]
+   ['subscription_name']
+
+Delete a subscription:
+
+.. doctest::
+
+   >>> from gcloud.pubsub.topic import Topic
+   >>> from gcloud.pubsub.subscription import Subscription
+   >>> topic = Topic('topic_name')
+   >>> subscription = Subscription('subscription_name', topic)
+   >>> subscription.delete()  # API request
+
+
+Pull messages from a subscription
+---------------------------------
+
+Fetch pending messages for a pull subscription
+
+.. note::
+
+   The messages will have been ACKed already.
+
+.. doctest::
+
+   >>> from gcloud.pubsub.topic import Topic
+   >>> from gcloud.pubsub.subscription import Subscription
+   >>> topic = Topic('topic_name')
+   >>> subscription = Subscription('subscription_name', topic)
+   >>> with topic:
+   ...     topic.publish('this is the first message_payload')
+   ...     topic.publish('this is the second message_payload',
+   ...                   attr1='value1', attr2='value2')
+   >>> messages = subscription.pull()  # API request
+   >>> [message.id for message in messages]
+   [<message_id1>, <message_id2>]
+   >>> [message.data for message in messages]
+   ['this is the first message_payload', 'this is the second message_payload']
+   >>> [message.attrs for message in messages]
+   [{}, {'attr1': 'value1', 'attr2': 'value2'}]
+
+Fetch a limited number of pending messages for a pull subscription:
+
+.. doctest::
+
+   >>> from gcloud.pubsub.topic import Topic
+   >>> from gcloud.pubsub.subscription import Subscription
+   >>> topic = Topic('topic_name')
+   >>> subscription = Subscription('subscription_name', topic)
+   >>> with topic:
+   ...     topic.publish('this is the first message_payload')
+   ...     topic.publish('this is the second message_payload',
+   ...                   attr1='value1', attr2='value2')
+   >>> [message.id for message in subscription.pull(max_messages=1)]
+   [<message_id1>]
+
+Fetch messages for a pull subscription without blocking (none pending):
+
+.. doctest::
+
+   >>> from gcloud.pubsub.topic import Topic
+   >>> from gcloud.pubsub.subscription import Subscription
+   >>> topic = Topic('topic_name')
+   >>> subscription = Subscription('subscription_name', topic)
+   >>> [message.id for message in subscription.pull(return_immediately=True)]
+   []
+

gcloud/pubsub/__init__.py

@@ -0,0 +1,15 @@
+# Copyright 2015 Google Inc. All rights reserved.
+#
+# 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.
+
+"""GCloud Pubsub API wrapper."""

gcloud/pubsub/api.py

@@ -0,0 +1,54 @@
+# Copyright 2015 Google Inc. All rights reserved.
+#
+# 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.
+
+""" Define API functions (not bound to classes)."""
+
+
+def list_topics(page_size=None, page_token=None,
+                project=None, connection=None):
+    """List topics for a given project.
+
+    :type page_size: int
+    :param page_size: maximum number of topics 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 topics. If not
+                       passed, the API will return the first page of topics.
+
+    :type project: string
+    :param project: project ID to query.  If not passed, defaults to the
+                    project ID inferred from the environment.
+
+    :type connection: :class:`gcloud.pubsub.connection.Connection`
+    :param connection: connection to use for the query.  If not passed,
+                       defaults to the connection inferred from the
+                       environment.
+
+    :rtype: dict
+    :returns: keys include ``topics`` (a list of topic mappings) and
+              ``nextPageToken`` (a string:  if non-empty, indicates that
+              more topics 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/topics' % project
+    return connection.api_request(method='GET', path=path, query_params=params)