diff --git a/docs/pubsub-subscription.rst b/docs/pubsub-subscription.rst index 43fc3344a918..6dda5fbdc593 100644 --- a/docs/pubsub-subscription.rst +++ b/docs/pubsub-subscription.rst @@ -3,5 +3,6 @@ Subscriptions .. automodule:: gcloud.pubsub.subscription :members: + :member-order: bysource :undoc-members: :show-inheritance: diff --git a/docs/pubsub-topic.rst b/docs/pubsub-topic.rst index 2b840b05db6a..99401f0cfbbe 100644 --- a/docs/pubsub-topic.rst +++ b/docs/pubsub-topic.rst @@ -3,5 +3,6 @@ Topics .. automodule:: gcloud.pubsub.topic :members: + :member-order: bysource :undoc-members: :show-inheritance: diff --git a/docs/pubsub-usage.rst b/docs/pubsub-usage.rst index 2f74fd584120..92eeee3e1281 100644 --- a/docs/pubsub-usage.rst +++ b/docs/pubsub-usage.rst @@ -28,84 +28,47 @@ Authentication / Configuration Manage topics for a project --------------------------- -Create a new topic for the default project: - -.. doctest:: +List topics for the default project: - >>> from gcloud import pubsub - >>> client = pubsub.Client() - >>> topic = client.topic('topic_name') - >>> topic.create() # API request +.. literalinclude:: pubsub_snippets.py + :start-after: [START client_list_topics] + :end-before: [END client_list_topics] -Check for the existence of a topic: - -.. doctest:: +Create a new topic for the default project: - >>> from gcloud import pubsub - >>> client = pubsub.Client() - >>> topic = client.topic('topic_name') - >>> topic.exists() # API request - True +.. literalinclude:: pubsub_snippets.py + :start-after: [START topic_create] + :end-before: [END topic_create] -List topics for the default project: - -.. doctest:: +Check for the existence of a topic: - >>> from gcloud import pubsub - >>> client = pubsub.Client() - >>> topics, next_page_token = client.list_topics() # API request - >>> [topic.name for topic in topics] - ['topic_name'] +.. literalinclude:: pubsub_snippets.py + :start-after: [START topic_exists] + :end-before: [END topic_exists] Delete a topic: -.. doctest:: - - >>> from gcloud import pubsub - >>> client = pubsub.Client() - >>> topic = client.topic('topic_name') - >>> topic.delete() # API request +.. literalinclude:: pubsub_snippets.py + :start-after: [START topic_delete] + :end-before: [END topic_delete] Fetch the IAM policy for a topic: -.. doctest:: - - >>> from gcloud import pubsub - >>> client = pubsub.Client() - >>> topic = client.topic('topic_name') - >>> policy = topic.get_iam_policy() # API request - >>> policy.etag - 'DEADBEEF' - >>> policy.owners - ['user:phred@example.com'] - >>> policy.writers - ['systemAccount:abc-1234@systemaccounts.example.com'] - >>> policy.readers - ['domain:example.com'] +.. literalinclude:: pubsub_snippets.py + :start-after: [START topic_get_iam_policy] + :end-before: [END topic_get_iam_policy] Update the IAM policy for a topic: -.. doctest:: - - >>> from gcloud import pubsub - >>> client = pubsub.Client() - >>> topic = client.topic('topic_name') - >>> policy = topic.get_iam_policy() # API request - >>> policy.writers.add(policy.group('editors-list@example.com')) - >>> topic.set_iam_policy(policy) # API request +.. literalinclude:: pubsub_snippets.py + :start-after: [START topic_set_iam_policy] + :end-before: [END topic_set_iam_policy] Test permissions allowed by the current IAM policy on a topic: -.. doctest:: - - >>> from gcloud import pubsub - >>> from gcloud.pubsub.iam import OWNER_ROLE, EDITOR_ROLE, VIEWER_ROLE - >>> client = pubsub.Client() - >>> topic = client.topic('topic_name') - >>> allowed = topic.check_iam_permissions( - ... [VIEWER_ROLE, EDITOR_ROLE, OWNER_ROLE]) # API request - >>> allowed == [VIEWER_ROLE, EDITOR_ROLE] - True +.. literalinclude:: pubsub_snippets.py + :start-after: [START topic_check_iam_permissions] + :end-before: [END topic_check_iam_permissions] Publish messages to a topic @@ -113,144 +76,93 @@ Publish messages to a topic Publish a single message to a topic, without attributes: -.. doctest:: - - >>> from gcloud import pubsub - >>> client = pubsub.Client() - >>> topic = client.topic('topic_name') - >>> topic.publish('this is the message_payload') # API request - +.. literalinclude:: pubsub_snippets.py + :start-after: [START topic_publish_simple_message] + :end-before: [END topic_publish_simple_message] Publish a single message to a topic, with attributes: -.. doctest:: - - >>> from gcloud import pubsub - >>> client = pubsub.Client() - >>> topic = client.topic('topic_name') - >>> topic.publish('this is another message_payload', - ... attr1='value1', attr2='value2') # API request - +.. literalinclude:: pubsub_snippets.py + :start-after: [START topic_publish_message_with_attrs] + :end-before: [END topic_publish_message_with_attrs] Publish a set of messages to a topic (as a single request): -.. doctest:: - - >>> from gcloud import pubsub - >>> client = pubsub.Client() - >>> topic = client.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) - [, ] +.. literalinclude:: pubsub_snippets.py + :start-after: [START topic_batch] + :end-before: [END topic_batch] .. note:: The only API request happens during the ``__exit__()`` of the topic - used as a context manager. + used as a context manager, and only if the block exits without raising + an exception. Manage subscriptions to topics ------------------------------ -Create a new pull subscription for a topic: +List all subscriptions for the default project: .. doctest:: - >>> from gcloud import pubsub - >>> client = pubsub.Client() - >>> topic = client.topic('topic_name') - >>> subscription = topic.subscription('subscription_name') - >>> subscription.create() # API request +.. literalinclude:: pubsub_snippets.py + :start-after: [START client_list_subscriptions] + :end-before: [END client_list_subscriptions] -Create a new pull subscription for a topic with a non-default ACK deadline: +List subscriptions for a topic: -.. doctest:: +.. literalinclude:: pubsub_snippets.py + :start-after: [START topic_list_subscriptions] + :end-before: [END topic_list_subscriptions] - >>> from gcloud import pubsub - >>> client = pubsub.Client() - >>> topic = client.topic('topic_name') - >>> subscription = topic.subscription('subscription_name', ack_deadline=90) - >>> subscription.create() # API request +Create a new pull subscription for a topic, with defaults: -Create a new push subscription for a topic: +.. literalinclude:: pubsub_snippets.py + :start-after: [START topic_subscription_defaults] + :end-before: [END topic_subscription_defaults] -.. doctest:: +Create a new pull subscription for a topic with a non-default ACK deadline: - >>> from gcloud import pubsub - >>> ENDPOINT = 'https://example.com/hook' - >>> client = pubsub.Client() - >>> topic = client.topic('topic_name') - >>> subscription = topic.subscription('subscription_name', - ... push_endpoint=ENDPOINT) - >>> subscription.create() # API request +.. literalinclude:: pubsub_snippets.py + :start-after: [START topic_subscription_ack90] + :end-before: [END topic_subscription_ack90] -Check for the existence of a subscription: +Create a new push subscription for a topic: -.. doctest:: +.. literalinclude:: pubsub_snippets.py + :start-after: [START topic_subscription_push] + :end-before: [END topic_subscription_push] - >>> from gcloud import pubsub - >>> client = pubsub.Client() - >>> topic = client.topic('topic_name') - >>> subscription = topic.subscription('subscription_name') - >>> subscription.exists() # API request - True +Check for the existence of a subscription: -Convert a pull subscription to push: +.. literalinclude:: pubsub_snippets.py + :start-after: [START subscription_exists] + :end-before: [END subscription_exists] -.. doctest:: +Convert a pull subscription to push: - >>> from gcloud import pubsub - >>> ENDPOINT = 'https://example.com/hook' - >>> client = pubsub.Client() - >>> topic = client.topic('topic_name') - >>> subscription = topic.subscription('subscription_name') - >>> subscription.modify_push_configuration(push_endpoint=ENDPOINT) # API request +.. literalinclude:: pubsub_snippets.py + :start-after: [START subscription_pull_push] + :end-before: [END subscription_pull_push] Convert a push subscription to pull: -.. doctest:: +.. literalinclude:: pubsub_snippets.py + :start-after: [START subscription_push_pull] + :end-before: [END subscription_push_pull] - >>> from gcloud import pubsub - >>> ENDPOINT = 'https://example.com/hook' - >>> client = pubsub.Client() - >>> topic = client.topic('topic_name') - >>> subscription = topic.subscription('subscription_name', - ... push_endpoint=ENDPOINT) - >>> subscription.modify_push_configuration(push_endpoint=None) # API request +Re-synchronize a subscription with the back-end: -List subscriptions for a topic: - -.. doctest:: - - >>> from gcloud import pubsub - >>> client = pubsub.Client() - >>> topic = client.topic('topic_name') - >>> subscriptions, next_page_token = topic.list_subscriptions() # API request - >>> [subscription.name for subscription in subscriptions] - ['subscription_name'] - -List all subscriptions for the default project: - -.. doctest:: - - >>> from gcloud import pubsub - >>> client = pubsub.Client() - >>> subscription, next_page_tokens = client.list_subscriptions() # API request - >>> [subscription.name for subscription in subscriptions] - ['subscription_name'] +.. literalinclude:: pubsub_snippets.py + :start-after: [START subscription_reload] + :end-before: [END subscription_reload] Delete a subscription: -.. doctest:: - - >>> from gcloud import pubsub - >>> client = pubsub.Client() - >>> topic = client.topic('topic_name') - >>> subscription = topic.subscription('subscription_name') - >>> subscription.delete() # API request +.. literalinclude:: pubsub_snippets.py + :start-after: [START subscription_delete] + :end-before: [END subscription_delete] Pull messages from a subscription @@ -258,102 +170,43 @@ Pull messages from a subscription Fetch pending messages for a pull subscription: -.. doctest:: - - >>> from gcloud import pubsub - >>> client = pubsub.Client() - >>> topic = client.topic('topic_name') - >>> subscription = topic.subscription('subscription_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') - >>> received = subscription.pull() # API request - >>> messages = [recv[1] for recv in received] - >>> [message.message_id for message in messages] - [, ] - >>> [message.data for message in messages] - ['this is the first message_payload', 'this is the second message_payload'] - >>> [message.attributes for message in messages] - [{}, {'attr1': 'value1', 'attr2': 'value2'}] +.. literalinclude:: pubsub_snippets.py + :start-after: [START subscription_pull] + :end-before: [END subscription_pull] Note that received messages must be acknowledged, or else the back-end will re-send them later: -.. doctest:: - - >>> ack_ids = [recv[0] for recv in received] - >>> subscription.acknowledge(ack_ids) - -Fetch a limited number of pending messages for a pull subscription: - -.. doctest:: - - >>> from gcloud import pubsub - >>> client = pubsub.Client() - >>> topic = client.topic('topic_name') - >>> subscription = topic.subscription('subscription_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') - >>> received = subscription.pull(max_messages=1) # API request - >>> messages = [recv[1] for recv in received] - >>> [message.message_id for message in messages] +.. literalinclude:: pubsub_snippets.py + :start-after: [START subscription_acknowledge] + :end-before: [END subscription_acknowledge] Fetch messages for a pull subscription without blocking (none pending): -.. doctest:: +.. literalinclude:: pubsub_snippets.py + :start-after: [START subscription_pull_return_immediately] + :end-before: [END subscription_pull_return_immediately] - >>> from gcloud import pubsub - >>> client = pubsub.Client() - >>> topic = client.topic('topic_name') - >>> subscription = topic.subscription('subscription_name') - >>> received = subscription.pull(return_immediately=True) # API request - >>> messages = [recv[1] for recv in received] - >>> [message.message_id for message in messages] - [] +Update the acknowlegement deadline for pulled messages: -Fetch the IAM policy for a subscription +.. literalinclude:: pubsub_snippets.py + :start-after: [START subscription_modify_ack_deadline] + :end-before: [END subscription_modify_ack_deadline] -.. doctest:: +Fetch the IAM policy for a subscription - >>> from gcloud import pubsub - >>> client = pubsub.Client() - >>> topic = client.topic('topic_name') - >>> subscription = topic.subscription('subscription_name') - >>> policy = subscription.get_iam_policy() # API request - >>> policy.etag - 'DEADBEEF' - >>> policy.owners - ['user:phred@example.com'] - >>> policy.writers - ['systemAccount:abc-1234@systemaccounts.example.com'] - >>> policy.readers - ['domain:example.com'] +.. literalinclude:: pubsub_snippets.py + :start-after: [START subscription_get_iam_policy] + :end-before: [END subscription_get_iam_policy] Update the IAM policy for a subscription: -.. doctest:: - - >>> from gcloud import pubsub - >>> client = pubsub.Client() - >>> topic = client.topic('topic_name') - >>> subscription = topic.subscription('subscription_name') - >>> policy = subscription.get_iam_policy() # API request - >>> policy.writers.add(policy.group('editors-list@example.com')) - >>> subscription.set_iam_policy(policy) # API request +.. literalinclude:: pubsub_snippets.py + :start-after: [START subscription_set_iam_policy] + :end-before: [END subscription_set_iam_policy] Test permissions allowed by the current IAM policy on a subscription: -.. doctest:: - - >>> from gcloud import pubsub - >>> from gcloud.pubsub.iam import OWNER_ROLE, EDITOR_ROLE, VIEWER_ROLE - >>> client = pubsub.Client() - >>> topic = client.topic('topic_name') - >>> subscription = topic.subscription('subscription_name') - >>> allowed = subscription.check_iam_permissions( - ... [VIEWER_ROLE, EDITOR_ROLE, OWNER_ROLE]) # API request - >>> allowed == [VIEWER_ROLE, EDITOR_ROLE] - True +.. literalinclude:: pubsub_snippets.py + :start-after: [START subscription_check_iam_permissions] + :end-before: [END subscription_check_iam_permissions] diff --git a/docs/pubsub_snippets.py b/docs/pubsub_snippets.py new file mode 100644 index 000000000000..6d932f56fe06 --- /dev/null +++ b/docs/pubsub_snippets.py @@ -0,0 +1,445 @@ +# Copyright 2016 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. + +"""Testable usage examples for Google Cloud Pubsub API wrapper + +Each example function takes a ``client`` argument (which must be an instance +of :class:`gcloud.pubsub.client.Client`) and uses it to perform a task with +the API. + +To facility running the examples as system tests, each example is also passed +a ``to_delete`` list; the function adds to the list any objects created which +need to be deleted during teardown. +""" + +import time + +from gcloud.pubsub.client import Client + + +def snippet(func): + """Mark ``func`` as a snippet example function.""" + func._snippet = True + return func + + +def _millis(): + return time.time() * 1000 + + +@snippet +def client_list_topics(client, to_delete): # pylint: disable=unused-argument + """List topics for a project.""" + + def do_something_with(sub): # pylint: disable=unused-argument + pass + + # [START client_list_topics] + topics, token = client.list_topics() # API request + while True: + for topic in topics: + do_something_with(topic) + if token is None: + break + topics, token = client.list_topics(page_token=token) # API request + # [END client_list_topics] + + +@snippet +def client_list_subscriptions(client, + to_delete): # pylint: disable=unused-argument + """List all subscriptions for a project.""" + + def do_something_with(sub): # pylint: disable=unused-argument + pass + + # [START client_list_subscriptions] + subscriptions, token = client.list_subscriptions() # API request + while True: + for subscription in subscriptions: + do_something_with(subscription) + if token is None: + break + subscriptions, token = client.list_subscriptions( + page_token=token) # API request + # [END client_list_subscriptions] + + +@snippet +def topic_create(client, to_delete): + """Create a topic.""" + TOPIC_NAME = 'topic_create-%d' % (_millis(),) + + # [START topic_create] + topic = client.topic(TOPIC_NAME) + topic.create() # API request + # [END topic_create] + + to_delete.append(topic) + + +@snippet +def topic_exists(client, to_delete): + """Test existence of a topic.""" + TOPIC_NAME = 'topic_exists-%d' % (_millis(),) + + # [START client_topic] + topic = client.topic(TOPIC_NAME) + # [END client_topic] + + to_delete.append(topic) + + # [START topic_exists] + assert not topic.exists() # API request + topic.create() # API request + assert topic.exists() # API request + # [END topic_exists] + + +@snippet +def topic_delete(client, to_delete): # pylint: disable=unused-argument + """Delete a topic.""" + TOPIC_NAME = 'topic_delete-%d' % (_millis(),) + topic = client.topic(TOPIC_NAME) + topic.create() # API request + + # [START topic_delete] + assert topic.exists() # API request + topic.delete() + assert not topic.exists() # API request + # [END topic_delete] + + +@snippet +def topic_iam_policy(client, to_delete): + """Fetch / set a topic's IAM policy.""" + TOPIC_NAME = 'topic_iam_policy-%d' % (_millis(),) + topic = client.topic(TOPIC_NAME) + topic.create() + to_delete.append(topic) + + # [START topic_get_iam_policy] + policy = topic.get_iam_policy() # API request + # [END topic_get_iam_policy] + + assert len(policy.viewers) == 0 + assert len(policy.editors) == 0 + assert len(policy.owners) == 0 + + # [START topic_set_iam_policy] + ALL_USERS = policy.all_users() + policy.viewers.add(ALL_USERS) + LOGS_GROUP = policy.group('cloud-logs@google.com') + policy.editors.add(LOGS_GROUP) + new_policy = topic.set_iam_policy(policy) # API request + # [END topic_set_iam_policy] + + assert ALL_USERS in new_policy.viewers + assert LOGS_GROUP in new_policy.editors + + +# @snippet # Disabled due to #1687 +def topic_check_iam_permissions(client, to_delete): + """Check topic IAM permissions.""" + TOPIC_NAME = 'topic_check_iam_permissions-%d' % (_millis(),) + topic = client.topic(TOPIC_NAME) + topic.create() + to_delete.append(topic) + + # [START topic_check_iam_permissions] + from gcloud.pubsub.iam import OWNER_ROLE, EDITOR_ROLE, VIEWER_ROLE + TO_CHECK = [OWNER_ROLE, EDITOR_ROLE, VIEWER_ROLE] + ALLOWED = topic.check_iam_permissions(TO_CHECK) + assert set(ALLOWED) == set(TO_CHECK) + # [END topic_check_iam_permissions] + + +@snippet +def topic_publish_messages(client, to_delete): + """Publish messages to a topic.""" + TOPIC_NAME = 'topic_publish_messages-%d' % (_millis(),) + topic = client.topic(TOPIC_NAME) + topic.create() + to_delete.append(topic) + + # [START topic_publish_simple_message] + topic.publish(b'This is the message payload') # API request + # [END topic_publish_simple_message] + + # [START topic_publish_message_with_attrs] + topic.publish(b'Another message payload', extra='EXTRA') # API request + # [END topic_publish_message_with_attrs] + + +@snippet +def topic_batch(client, to_delete): + """Publish multiple messages in a single request.""" + TOPIC_NAME = 'topic_batch-%d' % (_millis(),) + topic = client.topic(TOPIC_NAME) + topic.create() + to_delete.append(topic) + + # [START topic_batch] + with topic.batch() as batch: + batch.publish(b'This is the message payload') + batch.publish(b'Another message payload', extra='EXTRA') + # [END topic_batch] API request on block exit + + +@snippet +def topic_subscription(client, to_delete): + """Create subscriptions to a topic.""" + TOPIC_NAME = 'topic_subscription-%d' % (_millis(),) + SUB_DEFAULTS = 'topic_subscription-defaults-%d' % (_millis(),) + SUB_ACK90 = 'topic_subscription-ack90-%d' % (_millis(),) + topic = client.topic(TOPIC_NAME) + topic.create() + to_delete.append(topic) + + # [START topic_subscription_defaults] + sub_defaults = topic.subscription(SUB_DEFAULTS) + # [END topic_subscription_defaults] + + sub_defaults.create() # API request + to_delete.append(sub_defaults) + expected_names = set() + expected_names.add(sub_defaults.full_name) + + # [START topic_subscription_ack90] + sub_ack90 = topic.subscription(SUB_ACK90, ack_deadline=90) + # [END topic_subscription_ack90] + + sub_ack90.create() # API request + to_delete.append(sub_ack90) + expected_names.add(sub_ack90.full_name) + + sub_names = set() + + def do_something_with(sub): + sub_names.add(sub.full_name) + + # [START topic_list_subscriptions] + subscriptions, token = topic.list_subscriptions() # API request + while True: + for subscription in subscriptions: + do_something_with(subscription) + if token is None: + break + subscriptions, token = topic.list_subscriptions( + page_token=token) # API request + # [END topic_list_subscriptions] + + assert sub_names.issuperset(expected_names) + + +# @snippet: disabled, because push-mode requires a validated endpoint URL +def topic_subscription_push(client, to_delete): + """Create subscriptions to a topic.""" + TOPIC_NAME = 'topic_subscription_push-%d' % (_millis(),) + SUB_PUSH = 'topic_subscription_push-sub-%d' % (_millis(),) + PUSH_URL = 'https://api.example.com/push-endpoint' + topic = client.topic(TOPIC_NAME) + topic.create() + to_delete.append(topic) + + # [START topic_subscription_push] + subscription = topic.subscription(SUB_PUSH, push_endpoint=PUSH_URL) + subscription.create() # API request + # [END topic_subscription_push] + + # [START subscription_push_pull] + subscription.modify_push_configuration(push_endpoint=None) # API request + # [END subscription_push_pull] + + # [START subscription_pull_push] + subscription.modify_push_configuration( + push_endpoint=PUSH_URL) # API request + # [END subscription_pull_push] + + +@snippet +def subscription_lifecycle(client, to_delete): + """Test lifecycle of a subscription.""" + TOPIC_NAME = 'subscription_lifecycle-%d' % (_millis(),) + SUB_NAME = 'subscription_lifecycle-defaults-%d' % (_millis(),) + topic = client.topic(TOPIC_NAME) + topic.create() + to_delete.append(topic) + + # [START subscription_create] + subscription = topic.subscription(SUB_NAME) + subscription.create() # API request + # [END subscription_create] + + # [START subscription_exists] + assert subscription.exists() # API request + # [END subscription_exists] + + # [START subscription_reload] + subscription.reload() # API request + # [END subscription_reload] + + # [START subscription_delete] + subscription.delete() # API request + # [END subscription_delete] + + +@snippet +def subscription_pull(client, to_delete): + """Pull messges from a subscribed topic.""" + TOPIC_NAME = 'subscription_pull-%d' % (_millis(),) + SUB_NAME = 'subscription_pull-defaults-%d' % (_millis(),) + PAYLOAD1 = b'PAYLOAD1' + PAYLOAD2 = b'PAYLOAD2' + EXTRA = 'EXTRA' + topic = client.topic(TOPIC_NAME) + topic.create() + to_delete.append(topic) + + subscription = topic.subscription(SUB_NAME) + subscription.create() + to_delete.append(subscription) + + # [START subscription_pull_none_pending] + pulled = subscription.pull(max_messages=1) + # [END subscription_pull_none_pending] + assert len(pulled) == 0 + + # [START subscription_pull_return_immediately] + pulled = subscription.pull(return_immediately=True) + # [END subscription_pull_return_immediately] + assert len(pulled) == 0 + + topic.publish(PAYLOAD1) + topic.publish(PAYLOAD2, extra=EXTRA) + + # [START subscription_pull] + pulled = subscription.pull(max_messages=2) + # [END subscription_pull] + + assert len(pulled) == 2 + + # [START subscription_modify_ack_deadline] + for ack_id, _ in pulled: + subscription.modify_ack_deadline(ack_id, 90) # API request + # [END subscription_modify_ack_deadline] + + payloads = [] + extras = [] + + def do_something_with(message): # pylint: disable=unused-argument + payloads.append(message.data) + if message.attributes: + extras.append(message.attributes) + + class ApplicationException(Exception): + pass + + def log_exception(_): + pass + + # [START subscription_acknowledge] + for ack_id, message in pulled: + try: + do_something_with(message) + except ApplicationException as e: + log_exception(e) + else: + subscription.acknowledge([ack_id]) + # [END subscription_acknowledge] + + assert set(payloads) == set([PAYLOAD1, PAYLOAD2]), 'payloads: %s' % ( + (payloads,)) + assert extras == [{'extra': EXTRA}], 'extras: %s' % ( + (extras,)) + + +@snippet +def subscription_iam_policy(client, to_delete): + """Fetch / set a subscription's IAM policy.""" + TOPIC_NAME = 'subscription_iam_policy-%d' % (_millis(),) + SUB_NAME = 'subscription_iam_policy-defaults-%d' % (_millis(),) + topic = client.topic(TOPIC_NAME) + topic.create() + to_delete.append(topic) + + subscription = topic.subscription(SUB_NAME) + subscription.create() + to_delete.append(subscription) + + # [START subscription_get_iam_policy] + policy = subscription.get_iam_policy() # API request + # [END subscription_get_iam_policy] + + assert len(policy.viewers) == 0 + assert len(policy.editors) == 0 + assert len(policy.owners) == 0 + + # [START subscription_set_iam_policy] + ALL_USERS = policy.all_users() + policy.viewers.add(ALL_USERS) + LOGS_GROUP = policy.group('cloud-logs@google.com') + policy.editors.add(LOGS_GROUP) + new_policy = subscription.set_iam_policy(policy) # API request + # [END subscription_set_iam_policy] + + assert ALL_USERS in new_policy.viewers + assert LOGS_GROUP in new_policy.editors + + +# @snippet # Disabled due to #1687 +def subscription_check_iam_permissions(client, to_delete): + """Check subscription IAM permissions.""" + TOPIC_NAME = 'subscription_check_iam_permissions-%d' % (_millis(),) + SUB_NAME = 'subscription_check_iam_permissions-defaults-%d' % (_millis(),) + topic = client.topic(TOPIC_NAME) + topic.create() + to_delete.append(topic) + + subscription = topic.subscription(SUB_NAME) + subscription.create() + to_delete.append(subscription) + + # [START subscription_check_iam_permissions] + from gcloud.pubsub.iam import OWNER_ROLE, EDITOR_ROLE, VIEWER_ROLE + TO_CHECK = [OWNER_ROLE, EDITOR_ROLE, VIEWER_ROLE] + ALLOWED = subscription.check_iam_permissions(TO_CHECK) + assert set(ALLOWED) == set(TO_CHECK) + # [END subscription_check_iam_permissions] + + +def _find_examples(): + funcs = [obj for obj in globals().values() + if getattr(obj, '_snippet', False)] + for func in sorted(funcs, key=lambda f: f.func_code.co_firstlineno): + yield func + + +def main(): + client = Client() + for example in _find_examples(): + to_delete = [] + print('%-25s: %s' % ( + example.func_name, example.func_doc)) + try: + example(client, to_delete) + except AssertionError as e: + print(' FAIL: %s' % (e,)) + except Exception as e: # pylint: disable=broad-except + print(' ERROR: %r' % (e,)) + for item in to_delete: + item.delete() + +if __name__ == '__main__': + main() diff --git a/gcloud/pubsub/client.py b/gcloud/pubsub/client.py index 7f242eb2ee3b..d0c9e585539e 100644 --- a/gcloud/pubsub/client.py +++ b/gcloud/pubsub/client.py @@ -75,6 +75,12 @@ def list_topics(self, page_size=None, page_token=None): See: https://cloud.google.com/pubsub/reference/rest/v1/projects.topics/list + Example: + + .. literalinclude:: pubsub_snippets.py + :start-after: [START client_list_topics] + :end-before: [END client_list_topics] + :type page_size: int :param page_size: maximum number of topics to return, If not passed, defaults to a value set by the API. @@ -103,8 +109,11 @@ def list_subscriptions(self, page_size=None, page_token=None): See: https://cloud.google.com/pubsub/reference/rest/v1/projects.topics/list - and (where ``topic_name`` is passed): - https://cloud.google.com/pubsub/reference/rest/v1/projects.topics.subscriptions/list + Example: + + .. literalinclude:: pubsub_snippets.py + :start-after: [START client_list_subscriptions] + :end-before: [END client_list_subscriptions] :type page_size: int :param page_size: maximum number of topics to return, If not passed, @@ -133,6 +142,12 @@ def list_subscriptions(self, page_size=None, page_token=None): def topic(self, name, timestamp_messages=False): """Creates a topic bound to the current client. + Example: + + .. literalinclude:: pubsub_snippets.py + :start-after: [START client_topic] + :end-before: [END client_topic] + :type name: string :param name: the name of the topic to be constructed. diff --git a/gcloud/pubsub/subscription.py b/gcloud/pubsub/subscription.py index 5f01a9db8baf..8ce44e71cef4 100644 --- a/gcloud/pubsub/subscription.py +++ b/gcloud/pubsub/subscription.py @@ -115,9 +115,14 @@ def project(self): @property def full_name(self): - """URL path for the subscription's APIs""" + """Fully-qualified name used in subscription APIs""" return 'projects/%s/subscriptions/%s' % (self.project, self.name) + @property + def path(self): + """URL path for the subscription's APIs""" + return '/%s' % (self.full_name,) + def _require_client(self, client): """Check client or verify over-ride. @@ -139,6 +144,12 @@ def create(self, client=None): See: https://cloud.google.com/pubsub/reference/rest/v1/projects.subscriptions/create + Example: + + .. literalinclude:: pubsub_snippets.py + :start-after: [START subscription_create] + :end-before: [END subscription_create] + :type client: :class:`gcloud.pubsub.client.Client` or ``NoneType`` :param client: the client to use. If not passed, falls back to the ``client`` stored on the current subscription's topic. @@ -155,6 +166,12 @@ def exists(self, client=None): See https://cloud.google.com/pubsub/reference/rest/v1/projects.subscriptions/get + Example: + + .. literalinclude:: pubsub_snippets.py + :start-after: [START subscription_exists] + :end-before: [END subscription_exists] + :type client: :class:`gcloud.pubsub.client.Client` or ``NoneType`` :param client: the client to use. If not passed, falls back to the ``client`` stored on the current subscription's topic. @@ -174,6 +191,12 @@ def reload(self, client=None): See https://cloud.google.com/pubsub/reference/rest/v1/projects.subscriptions/get + Example: + + .. literalinclude:: pubsub_snippets.py + :start-after: [START subscription_reload] + :end-before: [END subscription_reload] + :type client: :class:`gcloud.pubsub.client.Client` or ``NoneType`` :param client: the client to use. If not passed, falls back to the ``client`` stored on the current subscription's topic. @@ -191,6 +214,12 @@ def delete(self, client=None): See: https://cloud.google.com/pubsub/reference/rest/v1/projects.subscriptions/delete + Example: + + .. literalinclude:: pubsub_snippets.py + :start-after: [START subscription_delete] + :end-before: [END subscription_delete] + :type client: :class:`gcloud.pubsub.client.Client` or ``NoneType`` :param client: the client to use. If not passed, falls back to the ``client`` stored on the current subscription's topic. @@ -205,6 +234,16 @@ def modify_push_configuration(self, push_endpoint, client=None): See: https://cloud.google.com/pubsub/reference/rest/v1/projects.subscriptions/modifyPushConfig + Example: + + .. literalinclude:: pubsub_snippets.py + :start-after: [START subscription_push_pull] + :end-before: [END subscription_push_pull] + + .. literalinclude:: pubsub_snippets.py + :start-after: [START subscription_pull_push] + :end-before: [END subscription_pull_push] + :type push_endpoint: string :param push_endpoint: URL to which messages will be pushed by the back-end. If None, the application must pull @@ -225,6 +264,12 @@ def pull(self, return_immediately=False, max_messages=1, client=None): See: https://cloud.google.com/pubsub/reference/rest/v1/projects.subscriptions/pull + Example: + + .. literalinclude:: pubsub_snippets.py + :start-after: [START subscription_pull] + :end-before: [END subscription_pull] + :type return_immediately: boolean :param return_immediately: if True, the back-end returns even if no messages are available; if False, the API @@ -256,6 +301,12 @@ def acknowledge(self, ack_ids, client=None): See: https://cloud.google.com/pubsub/reference/rest/v1/projects.subscriptions/acknowledge + Example: + + .. literalinclude:: pubsub_snippets.py + :start-after: [START subscription_acknowledge] + :end-before: [END subscription_acknowledge] + :type ack_ids: list of string :param ack_ids: ack IDs of messages being acknowledged @@ -294,6 +345,12 @@ def get_iam_policy(self, client=None): See: https://cloud.google.com/pubsub/reference/rest/v1/projects.subscriptions/getIamPolicy + Example: + + .. literalinclude:: pubsub_snippets.py + :start-after: [START subscription_get_iam_policy] + :end-before: [END subscription_get_iam_policy] + :type client: :class:`gcloud.pubsub.client.Client` or ``NoneType`` :param client: the client to use. If not passed, falls back to the ``client`` stored on the current subscription's topic. @@ -313,6 +370,12 @@ def set_iam_policy(self, policy, client=None): See: https://cloud.google.com/pubsub/reference/rest/v1/projects.subscriptions/setIamPolicy + Example: + + .. literalinclude:: pubsub_snippets.py + :start-after: [START subscription_set_iam_policy] + :end-before: [END subscription_set_iam_policy] + :type policy: :class:`gcloud.pubsub.iam.Policy` :param policy: the new policy, typically fetched via :meth:`get_iam_policy` and updated in place. @@ -337,6 +400,12 @@ def check_iam_permissions(self, permissions, client=None): See: https://cloud.google.com/pubsub/reference/rest/v1/projects.subscriptions/testIamPermissions + Example: + + .. literalinclude:: pubsub_snippets.py + :start-after: [START subscription_check_iam_permissions] + :end-before: [END subscription_check_iam_permissions] + :type permissions: list of string :param permissions: list of permissions to be tested diff --git a/gcloud/pubsub/test_subscription.py b/gcloud/pubsub/test_subscription.py index fdcc03aa4d50..fa1bd0f19696 100644 --- a/gcloud/pubsub/test_subscription.py +++ b/gcloud/pubsub/test_subscription.py @@ -132,6 +132,18 @@ def test_from_api_repr_w_topics_w_topic_match(self): self.assertEqual(subscription.ack_deadline, self.DEADLINE) self.assertEqual(subscription.push_endpoint, self.ENDPOINT) + def test_full_name_and_path(self): + PROJECT = 'PROJECT' + SUB_NAME = 'sub_name' + SUB_FULL = 'projects/%s/subscriptions/%s' % (PROJECT, SUB_NAME) + SUB_PATH = '/%s' % (SUB_FULL,) + TOPIC_NAME = 'topic_name' + CLIENT = _Client(project=PROJECT) + topic = _Topic(TOPIC_NAME, client=CLIENT) + subscription = self._makeOne(SUB_NAME, topic) + self.assertEqual(subscription.full_name, SUB_FULL) + self.assertEqual(subscription.path, SUB_PATH) + def test_create_pull_wo_ack_deadline_w_bound_client(self): RESPONSE = { 'topic': self.TOPIC_PATH, diff --git a/gcloud/pubsub/topic.py b/gcloud/pubsub/topic.py index 1e01d52f221b..6169ada84911 100644 --- a/gcloud/pubsub/topic.py +++ b/gcloud/pubsub/topic.py @@ -53,6 +53,24 @@ def __init__(self, name, client, timestamp_messages=False): def subscription(self, name, ack_deadline=None, push_endpoint=None): """Creates a subscription bound to the current topic. + Example: pull-mode subcription, default paramter values + + .. literalinclude:: pubsub_snippets.py + :start-after: [START topic_subscription_defaults] + :end-before: [END topic_subscription_defaults] + + Example: pull-mode subcription, override ``ack_deadline`` default + + .. literalinclude:: pubsub_snippets.py + :start-after: [START topic_subscription_ack90] + :end-before: [END topic_subscription_ack90] + + Example: push-mode subcription + + .. literalinclude:: pubsub_snippets.py + :start-after: [START topic_subscription_push] + :end-before: [END topic_subscription_push] + :type name: string :param name: the name of the subscription @@ -118,6 +136,12 @@ def create(self, client=None): See: https://cloud.google.com/pubsub/reference/rest/v1/projects.topics/create + Example: + + .. literalinclude:: pubsub_snippets.py + :start-after: [START topic_create] + :end-before: [END topic_create] + :type client: :class:`gcloud.pubsub.client.Client` or ``NoneType`` :param client: the client to use. If not passed, falls back to the ``client`` stored on the current topic. @@ -132,6 +156,12 @@ def exists(self, client=None): See https://cloud.google.com/pubsub/reference/rest/v1/projects.topics/get + Example: + + .. literalinclude:: pubsub_snippets.py + :start-after: [START topic_exists] + :end-before: [END topic_exists] + :type client: :class:`gcloud.pubsub.client.Client` or ``NoneType`` :param client: the client to use. If not passed, falls back to the ``client`` stored on the current topic. @@ -152,6 +182,12 @@ def delete(self, client=None): See: https://cloud.google.com/pubsub/reference/rest/v1/projects.topics/delete + Example: + + .. literalinclude:: pubsub_snippets.py + :start-after: [START topic_delete] + :end-before: [END topic_delete] + :type client: :class:`gcloud.pubsub.client.Client` or ``NoneType`` :param client: the client to use. If not passed, falls back to the ``client`` stored on the current topic. @@ -176,6 +212,18 @@ def publish(self, message, client=None, **attrs): See: https://cloud.google.com/pubsub/reference/rest/v1/projects.topics/publish + Example without message attributes: + + .. literalinclude:: pubsub_snippets.py + :start-after: [START topic_publish_simple_message] + :end-before: [END topic_publish_simple_message] + + With message attributes: + + .. literalinclude:: pubsub_snippets.py + :start-after: [START topic_publish_message_with_attrs] + :end-before: [END topic_publish_message_with_attrs] + :type message: bytes :param message: the message payload @@ -201,6 +249,18 @@ def publish(self, message, client=None, **attrs): def batch(self, client=None): """Return a batch to use as a context manager. + Example: + + .. literalinclude:: pubsub_snippets.py + :start-after: [START topic_batch] + :end-before: [END topic_batch] + + .. note:: + + The only API request happens during the ``__exit__()`` of the topic + used as a context manager, and only if the block exits without + raising an exception. + :type client: :class:`gcloud.pubsub.client.Client` or ``NoneType`` :param client: the client to use. If not passed, falls back to the ``client`` stored on the current topic. @@ -217,6 +277,12 @@ def list_subscriptions(self, page_size=None, page_token=None, client=None): See: https://cloud.google.com/pubsub/reference/rest/v1/projects.topics.subscriptions/list + Example: + + .. literalinclude:: pubsub_snippets.py + :start-after: [START topic_list_subscriptions] + :end-before: [END topic_list_subscriptions] + :type page_size: int :param page_size: maximum number of topics to return, If not passed, defaults to a value set by the API. @@ -252,6 +318,12 @@ def get_iam_policy(self, client=None): See: https://cloud.google.com/pubsub/reference/rest/v1/projects.topics/getIamPolicy + Example: + + .. literalinclude:: pubsub_snippets.py + :start-after: [START topic_get_iam_policy] + :end-before: [END topic_get_iam_policy] + :type client: :class:`gcloud.pubsub.client.Client` or ``NoneType`` :param client: the client to use. If not passed, falls back to the ``client`` stored on the current batch. @@ -271,6 +343,12 @@ def set_iam_policy(self, policy, client=None): See: https://cloud.google.com/pubsub/reference/rest/v1/projects.topics/setIamPolicy + Example: + + .. literalinclude:: pubsub_snippets.py + :start-after: [START topic_set_iam_policy] + :end-before: [END topic_set_iam_policy] + :type policy: :class:`gcloud.pubsub.iam.Policy` :param policy: the new policy, typically fetched via :meth:`get_iam_policy` and updated in place. @@ -295,6 +373,12 @@ def check_iam_permissions(self, permissions, client=None): See: https://cloud.google.com/pubsub/reference/rest/v1/projects.topics/testIamPermissions + Example: + + .. literalinclude:: pubsub_snippets.py + :start-after: [START topic_check_iam_permissions] + :end-before: [END topic_check_iam_permissions] + :type permissions: list of string :param permissions: list of permissions to be tested diff --git a/scripts/run_pylint.py b/scripts/run_pylint.py index fe5c772d4c5d..113236640af4 100644 --- a/scripts/run_pylint.py +++ b/scripts/run_pylint.py @@ -125,7 +125,7 @@ def is_production_filename(filename): :rtype: bool :returns: Boolean indicating production status. """ - return 'test' not in filename + return 'test' not in filename and 'docs' not in filename def get_files_for_linting(allow_limited=True):