Repository: mercadona/rele
Pull request #287 (2025-01-24 15:24:54)

Allow create topics on multiple Google Pub/Sub regions
### :tophat: What?

Allow creating topics on multiple Google Pub/Sub regions

### :thinking: Why?

We want topics to be created on multiple Google Pub/Sub regions to improve reliability and GDPR compliance

### :link: Related issue
#288 



Patch:
diff --git a/docs/settings.rst b/docs/settings.rst
index 5e6e10a..bd70dcb 100644
--- a/docs/settings.rst
+++ b/docs/settings.rst
@@ -31,7 +31,7 @@ Example::
         'PUBLISHER_TIMEOUT': 3.0,
         'FILTER_SUBS_BY': boolean_function,
         'DEFAULT_RETRY_POLICY': RetryPolicy(10, 50),
-        'GC_STORAGE_REGION': 'europe-west1',
+        'GC_STORAGE_REGION': ["europe-southwest1", "europe-west1", "europe-west8", "europe-west9"],
         'CLIENT_OPTIONS': {'api_endpoint': 'custom-api.interconnect.example.com'}
     }
 
@@ -195,7 +195,7 @@ RetryPolicy will be triggered on NACKs or acknowledgement deadline exceeded even
 
 **Optional**
 
-Set the Google Cloud's region for storing the messages. By default is `europe-west1`
+Set the Google Cloud's region for storing the messages. By default is `["europe-southwest1", "europe-west1", "europe-west8", "europe-west9"]`
 
 ``CLIENT_OPTIONS``
 ----------------------------
diff --git a/rele/client.py b/rele/client.py
index 019fc8b..e7d0d6a 100644
--- a/rele/client.py
+++ b/rele/client.py
@@ -2,6 +2,7 @@ import json
 import logging
 import os
 import time
+import warnings
 from concurrent.futures import TimeoutError
 
 import google.auth
@@ -55,7 +56,9 @@ class Subscriber:
         self._gc_project_id = gc_project_id
         self._ack_deadline = default_ack_deadline or DEFAULT_ACK_DEADLINE
         self.credentials = credentials if not USE_EMULATOR else None
-        self._message_storage_policy = message_storage_policy
+        self._message_storage_policy = self._normalize_storage_policy(
+            message_storage_policy
+        )
         self._client = pubsub_v1.SubscriberClient(
             credentials=credentials, client_options=client_options
         )
@@ -95,11 +98,34 @@ class Subscriber:
             request={
                 "name": topic_path,
                 "message_storage_policy": MessageStoragePolicy(
-                    {"allowed_persistence_regions": [self._message_storage_policy]}
+                    {"allowed_persistence_regions": self._message_storage_policy}
                 ),
             }
         )
 
+    def _normalize_storage_policy(self, policy):
+        """
+        Normalizes the storage policy to accept either a string or a list.
+        If it is a string, it converts it into a list with a single element.
+        """
+
+        if isinstance(policy, str):
+            # Warning log for future compatibility
+            warnings.warn(
+                "`message_storage_policy` as a string is deprecated. Use a list of regions instead.",
+                DeprecationWarning,
+            )
+            return [policy]
+        if isinstance(policy, list):
+            if not policy:
+                raise ValueError("The storage policy must include at least one region.")
+            return policy
+        if policy is None:
+            return policy
+        raise TypeError(
+            "`message_storage_policy` must be None or either a string or a list of regions."
+        )
+
     def _create_subscription(self, subscription_path, topic_path, subscription):
         request = {
             "name": subscription_path,
diff --git a/rele/config.py b/rele/config.py
index c3eaecd..7b9a4ce 100644
--- a/rele/config.py
+++ b/rele/config.py
@@ -28,7 +28,10 @@ class Config:
     def __init__(self, setting):
         self._gc_project_id = setting.get("GC_PROJECT_ID")
         self.gc_credentials_path = setting.get("GC_CREDENTIALS_PATH")
-        self.gc_storage_region = setting.get("GC_STORAGE_REGION", "europe-west1")
+        self.gc_storage_region = setting.get(
+            "GC_STORAGE_REGION",
+            ["europe-southwest1", "europe-west1", "europe-west8", "europe-west9"],
+        )
         self.app_name = setting.get("APP_NAME")
         self.sub_prefix = setting.get("SUB_PREFIX")
         self.middleware = setting.get("MIDDLEWARE", default_middleware)


Test patch:
diff --git a/tests/commands/test_runrele.py b/tests/commands/test_runrele.py
index 357b2b3..fa6485b 100644
--- a/tests/commands/test_runrele.py
+++ b/tests/commands/test_runrele.py
@@ -18,7 +18,14 @@ class TestRunReleCommand:
         call_command("runrele")
 
         mock_worker.assert_called_with(
-            [], None, "rele-test", ANY, "europe-west1", 60, 2, None
+            [],
+            None,
+            "rele-test",
+            ANY,
+            ["europe-southwest1", "europe-west1", "europe-west8", "europe-west9"],
+            60,
+            2,
+            None,
         )
         mock_worker.return_value.run_forever.assert_called_once_with()
 
@@ -35,6 +42,13 @@ class TestRunReleCommand:
             "be exhausted." in err
         )
         mock_worker.assert_called_with(
-            [], None, "rele-test", ANY, "europe-west1", 60, 2, None
+            [],
+            None,
+            "rele-test",
+            ANY,
+            ["europe-southwest1", "europe-west1", "europe-west8", "europe-west9"],
+            60,
+            2,
+            None,
         )
         mock_worker.return_value.run_forever.assert_called_once_with()
diff --git a/tests/conftest.py b/tests/conftest.py
index 71f5f58..35c163a 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -34,6 +34,20 @@ def config(project_id):
     )
 
 
+@pytest.fixture
+def config_with_multiple_regions(project_id):
+    return Config(
+        {
+            "APP_NAME": "rele",
+            "SUB_PREFIX": "rele",
+            "GC_CREDENTIALS_PATH": "tests/dummy-pub-sub-credentials.json",
+            "GC_STORAGE_REGION": ["some-region", "another-region"],
+            "MIDDLEWARE": ["rele.contrib.LoggingMiddleware"],
+            "CLIENT_OPTIONS": {"api_endpoint": "custom-api.interconnect.example.com"},
+        }
+    )
+
+
 @pytest.fixture
 def config_with_retry_policy(project_id):
     return Config(
@@ -65,6 +79,17 @@ def subscriber(project_id, config):
     )
 
 
+@pytest.fixture
+def subscriber_with_multiple_storage_regions(project_id, config_with_multiple_regions):
+    return Subscriber(
+        config_with_multiple_regions.gc_project_id,
+        config_with_multiple_regions.credentials,
+        config_with_multiple_regions.gc_storage_region,
+        config_with_multiple_regions.client_options,
+        60,
+    )
+
+
 @pytest.fixture
 def mock_future():
     return MagicMock(spec=concurrent.futures.Future)
diff --git a/tests/test_subscriber.py b/tests/test_subscriber.py
index a65ab37..1a17923 100644
--- a/tests/test_subscriber.py
+++ b/tests/test_subscriber.py
@@ -166,6 +166,51 @@ class TestSubscriber:
             }
         )
 
+    @patch.object(
+        SubscriberClient,
+        "create_subscription",
+        side_effect=[exceptions.NotFound("Subscription topic does not exist"), True],
+    )
+    @patch.object(SubscriberClient, "update_subscription")
+    def test_creates_topic_when_subscription_topic_does_not_exist_in_multiple_regions(
+        self,
+        client_update_subscription,
+        client_create_subscription,
+        project_id,
+        subscriber_with_multiple_storage_regions,
+        mock_create_topic,
+    ):
+        expected_subscription = (
+            f"projects/{project_id}/subscriptions/" f"{project_id}-test-topic"
+        )
+        expected_topic = f"projects/{project_id}/topics/" f"{project_id}-test-topic"
+        backend_filter_by = "attributes:domain"
+        subscriber_with_multiple_storage_regions.update_or_create_subscription(
+            Subscription(
+                None,
+                topic=f"{project_id}-test-topic",
+                backend_filter_by=backend_filter_by,
+            )
+        )
+
+        mock_create_topic.assert_called_with(
+            request={
+                "name": f"projects/rele-test/topics/{project_id}-test-topic",
+                "message_storage_policy": MessageStoragePolicy(
+                    {"allowed_persistence_regions": ["some-region", "another-region"]}
+                ),
+            }
+        )
+
+        client_create_subscription.assert_called_with(
+            request={
+                "ack_deadline_seconds": 60,
+                "name": expected_subscription,
+                "topic": expected_topic,
+                "filter": backend_filter_by,
+            }
+        )
+
     @patch.object(SubscriberClient, "create_subscription")
     @patch.object(SubscriberClient, "update_subscription")
     def test_creates_subscription_with_retry_policy_when_provided(
diff --git a/tests/test_worker.py b/tests/test_worker.py
index a753787..3ce174b 100644
--- a/tests/test_worker.py
+++ b/tests/test_worker.py
@@ -133,8 +133,8 @@ class TestWorker:
             subscriptions,
             config.client_options,
             config.gc_project_id,
-            config.gc_storage_region,
             config.credentials,
+            config.gc_storage_region,
             custom_ack_deadline,
             threads_per_subscription=10,
         )