Repository: mercadona/rele
Pull request #206 (2021-09-07 07:38:03)

Automatic topic creation
### :tophat: What?

Automatic topic creation when topic does not exist in PubSub.

### :thinking: Why?

When we trying to subscribe to a topic that doesn't exist, there is a crash. In this feature, we create the topic automatically if this does not exist and then retry to initialize the subscription.

PubSub credentials must have editor role.

### :link: Related issue

None



Patch:
diff --git a/rele/client.py b/rele/client.py
index a500582..146e348 100644
--- a/rele/client.py
+++ b/rele/client.py
@@ -40,11 +40,8 @@ class Subscriber:
     def __init__(self, gc_project_id, credentials, default_ack_deadline=None):
         self._gc_project_id = gc_project_id
         self._ack_deadline = default_ack_deadline or DEFAULT_ACK_DEADLINE
-
-        if USE_EMULATOR:
-            self._client = pubsub_v1.SubscriberClient()
-        else:
-            self._client = pubsub_v1.SubscriberClient(credentials=credentials)
+        self.credentials = credentials if not USE_EMULATOR else None
+        self._client = pubsub_v1.SubscriberClient(credentials=credentials)
 
     def create_subscription(self, subscription, topic):
         """Handles creating the subscription when it does not exists.
@@ -64,13 +61,26 @@ class Subscriber:
 
         with suppress(exceptions.AlreadyExists):
             try:
-                self._client.create_subscription(
-                    name=subscription_path,
-                    topic=topic_path,
-                    ack_deadline_seconds=self._ack_deadline,
-                )
+                self._create_subscription(subscription_path, topic_path)
             except exceptions.NotFound:
-                logger.error("Cannot subscribe to a topic that does not exist.")
+                logger.warning(
+                    "Cannot subscribe to a topic that does not exist."
+                    f"Creating {topic_path}..."
+                )
+                topic = self._create_topic(topic_path)
+                logger.info(f"Topic {topic.name} created.")
+                self._create_subscription(subscription_path, topic_path)
+
+    def _create_topic(self, topic_path):
+        publisher_client = pubsub_v1.PublisherClient(credentials=self.credentials)
+        return publisher_client.create_topic(request={"name": topic_path})
+
+    def _create_subscription(self, name, topic):
+        self._client.create_subscription(
+            name=name,
+            topic=topic,
+            ack_deadline_seconds=self._ack_deadline,
+        )
 
     def consume(self, subscription_name, callback, scheduler):
         """Begin listening to topic from the SubscriberClient.


Test patch:
diff --git a/tests/test_client.py b/tests/test_client.py
index 12c4355..c88969c 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -5,7 +5,7 @@ from unittest.mock import ANY, patch
 
 import pytest
 from google.api_core import exceptions
-from google.cloud.pubsub_v1 import SubscriberClient
+from google.cloud.pubsub_v1 import PublisherClient, SubscriberClient
 from google.cloud.pubsub_v1.exceptions import TimeoutError
 
 
@@ -135,6 +135,13 @@ class TestPublisher:
 
 
 class TestSubscriber:
+    @pytest.fixture(autouse=True)
+    def mock_create_topic(self):
+        with patch.object(
+            PublisherClient, "create_topic", return_values={"name": "test-topic"}
+        ) as mock:
+            yield mock
+
     @patch.object(SubscriberClient, "create_subscription")
     def test_creates_subscription_with_default_ack_deadline_when_none_provided(
         self,
@@ -187,16 +194,16 @@ class TestSubscriber:
     @patch.object(
         SubscriberClient,
         "create_subscription",
-        side_effect=exceptions.NotFound("Subscription topic does not exist"),
+        side_effect=[exceptions.NotFound("Subscription topic does not exist"), True],
     )
-    def test_logs_error_when_subscription_topic_does_not_exist(
-        self, _mocked_client, project_id, subscriber, caplog
+    def test_creates_topic_when_subscription_topic_does_not_exist(
+        self, _mocked_client, project_id, subscriber, mock_create_topic
     ):
         subscriber.create_subscription(
             subscription="test-topic", topic=f"{project_id}-test-topic"
         )
 
-        _mocked_client.assert_called()
-        log = caplog.records[0]
-        assert log.message == "Cannot subscribe to a topic that does not exist."
-        assert log.levelname == "ERROR"
+        assert _mocked_client.call_count == 2
+        mock_create_topic.assert_called_with(
+            request={"name": f"projects/rele-test/topics/{project_id}-test-topic"}
+        )