Expunge 'Topic.connection' property.

Adopt '_require_connection' pattern from storage.

Addresses:
#859 (comment)

Author: tseaver

gcloud/pubsub/api.py

@@ -15,7 +15,7 @@
 """Define API functions (not bound to classes)."""
 
 from gcloud._helpers import get_default_project
-from gcloud.pubsub._implicit_environ import get_default_connection
+from gcloud.pubsub._implicit_environ import _require_connection
 from gcloud.pubsub.subscription import Subscription
 from gcloud.pubsub.topic import Topic
 
@@ -53,8 +53,7 @@ def list_topics(page_size=None, page_token=None,
     if project is None:
         project = get_default_project()
 
-    if connection is None:
-        connection = get_default_connection()
+    connection = _require_connection(connection)
 
     params = {}
 
@@ -66,8 +65,7 @@ def list_topics(page_size=None, page_token=None,
 
     path = '/projects/%s/topics' % project
     resp = connection.api_request(method='GET', path=path, query_params=params)
-    topics = [Topic.from_api_repr(resource, connection)
-              for resource in resp['topics']]
+    topics = [Topic.from_api_repr(resource) for resource in resp['topics']]
     return topics, resp.get('nextPageToken')
 
 
@@ -110,8 +108,7 @@ def list_subscriptions(page_size=None, page_token=None, topic_name=None,
     if project is None:
         project = get_default_project()
 
-    if connection is None:
-        connection = get_default_connection()
+    connection = _require_connection(connection)
 
     params = {}
 
@@ -128,8 +125,6 @@ def list_subscriptions(page_size=None, page_token=None, topic_name=None,
 
     resp = connection.api_request(method='GET', path=path, query_params=params)
     topics = {}
-    subscriptions = [Subscription.from_api_repr(resource,
-                                                connection=connection,
-                                                topics=topics)
+    subscriptions = [Subscription.from_api_repr(resource, topics=topics)
                      for resource in resp['subscriptions']]
     return subscriptions, resp.get('nextPageToken')

gcloud/pubsub/subscription.py

@@ -17,6 +17,7 @@
 from gcloud.exceptions import NotFound
 from gcloud.pubsub.message import Message
 from gcloud.pubsub.topic import Topic
+from gcloud.pubsub._implicit_environ import _require_connection
 
 
 class Subscription(object):
@@ -46,17 +47,12 @@ def __init__(self, name, topic, ack_deadline=None, push_endpoint=None):
         self.push_endpoint = push_endpoint
 
     @classmethod
-    def from_api_repr(cls, resource, connection=None, topics=None):
+    def from_api_repr(cls, resource, topics=None):
         """Factory:  construct a topic given its API representation
 
         :type resource: dict
         :param resource: topic resource representation returned from the API
 
-        :type connection: :class:`gcloud.pubsub.connection.Connection` or None
-        :param connection: the connection to use.  If not passed,
-                           falls back to the default inferred from the
-                           environment.
-
         :type topics: dict or None
         :param topics: A mapping of topic names -> topics.  If not passed,
                        the subscription will have a newly-created topic.
@@ -68,8 +64,7 @@ def from_api_repr(cls, resource, connection=None, topics=None):
         t_name = resource['topic']
         topic = topics.get(t_name)
         if topic is None:
-            topic = topics[t_name] = Topic.from_api_repr({'name': t_name},
-                                                         connection)
+            topic = topics[t_name] = Topic.from_api_repr({'name': t_name})
         _, _, _, name = resource['name'].split('/')
         ack_deadline = resource.get('ackDeadlineSeconds')
         push_config = resource.get('pushConfig', {})
@@ -100,9 +95,7 @@ def create(self, connection=None):
         if self.push_endpoint is not None:
             data['pushConfig'] = {'pushEndpoint': self.push_endpoint}
 
-        if connection is None:
-            connection = self.topic.connection
-
+        connection = _require_connection(connection)
         connection.api_request(method='PUT', path=self.path, data=data)
 
     def exists(self, connection=None):
@@ -115,8 +108,7 @@ def exists(self, connection=None):
         :param connection: the connection to use.  If not passed,
                            falls back to the topic's connection.
         """
-        if connection is None:
-            connection = self.topic.connection
+        connection = _require_connection(connection)
         try:
             connection.api_request(method='GET', path=self.path)
         except NotFound:
@@ -134,8 +126,7 @@ def reload(self, connection=None):
         :param connection: the connection to use.  If not passed,
                            falls back to the topic's connection.
         """
-        if connection is None:
-            connection = self.topic.connection
+        connection = _require_connection(connection)
         data = connection.api_request(method='GET', path=self.path)
         self.ack_deadline = data.get('ackDeadline')
         push_config = data.get('pushConfig', {})
@@ -156,8 +147,7 @@ def modify_push_configuration(self, push_endpoint, connection=None):
         :param connection: the connection to use.  If not passed,
                            falls back to the topic's connection.
         """
-        if connection is None:
-            connection = self.topic.connection
+        connection = _require_connection(connection)
         data = {}
         config = data['pushConfig'] = {}
         if push_endpoint is not None:
@@ -191,8 +181,7 @@ def pull(self, return_immediately=False, max_messages=1, connection=None):
                   subsequent call to :meth:`acknowledge`, and ``message``
                   is an instance of :class:`gcloud.pubsub.message.Message`.
         """
-        if connection is None:
-            connection = self.topic.connection
+        connection = _require_connection(connection)
         data = {'returnImmediately': return_immediately,
                 'maxMessages': max_messages}
         response = connection.api_request(method='POST',
@@ -214,8 +203,7 @@ def acknowledge(self, ack_ids, connection=None):
         :param connection: the connection to use.  If not passed,
                            falls back to the topic's connection.
         """
-        if connection is None:
-            connection = self.topic.connection
+        connection = _require_connection(connection)
         data = {'ackIds': ack_ids}
         connection.api_request(method='POST',
                                path='%s:acknowledge' % self.path,
@@ -237,8 +225,7 @@ def modify_ack_deadline(self, ack_id, ack_deadline, connection=None):
         :param connection: the connection to use.  If not passed,
                            falls back to the topic's connection.
         """
-        if connection is None:
-            connection = self.topic.connection
+        connection = _require_connection(connection)
         data = {'ackId': ack_id, 'ackDeadlineSeconds': ack_deadline}
         connection.api_request(method='POST',
                                path='%s:modifyAckDeadline' % self.path,
@@ -254,6 +241,5 @@ def delete(self, connection=None):
         :param connection: the connection to use.  If not passed,
                            falls back to the topic's connection.
         """
-        if connection is None:
-            connection = self.topic.connection
+        connection = _require_connection(connection)
         connection.api_request(method='DELETE', path=self.path)