Repository: mercadona/rele
Pull request #172 (2020-06-28 18:28:02)

Catch TimeoutError and run post_publish_failure when blocking
### :tophat: What?

Catch `TimeoutError` when publishing while `blocking` and run `post_publish_failure` hook. Also added `raise_exception` parameter.
### :thinking: Why?
Google can raise this error when publishing so it should be handled
### :link: Related issue
#159 



Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.rst
similarity index 61%
rename from CHANGELOG.md
rename to CHANGELOG.rst
index 077a801..5457079 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.rst
@@ -1,122 +1,142 @@
 Changelog
 =========
 
-`0.13.0` (2020-7-9)
+0.13.0 (2020-7-9)
+-------------------
 * Add documentation for class based subscriptions (#169)
 * Deprecate GC_CREDENTIALS setting (#173)
 * GC_CREDENTIALS_PATH setting option (#170)
 
-`0.13.dev0` (2020-6-16)
+0.13.dev0 (2020-6-16)
+---------------------
 * Traverse all packages to autodiscover all subs.py modules (#167)
 * Auto-discovery of class based subscriptions (#168)
 
-`0.12.0` (2020-6-12)
-* Added `--settings` path option in CLI (#166)
+0.12.0 (2020-6-12)
+-------------------
+* Added ``--settings`` path option in CLI (#166)
 * Added isort linting (#164)
 
-`0.11.0` (2020-6-4)
+0.11.0 (2020-6-4)
+-------------------
 * CLI feature (#160)
 * Documentation Enhancements (#158, #155, #162)
 * Testing Improvements (#154, #157)
 
-`0.10.0` (2020-2-4)
+0.10.0 (2020-2-4)
+-------------------
 * Adjust default THREADS_PER_SUBSCRIPTION (#152)
 * Add unrecoverable_middleware (#150)
 * Allow multiple filters (#148)
 * Configure timeout from .publish() (#143)
 * Dont crash when subscription topic does not exist (#142)
 
-`0.9.1` (2020-1-2)
+0.9.1 (2020-1-2)
+-------------------
 * Ack messages when data not json serializable (#141)
 * Use ThreadScheduler instead of ThreadPoolExecutor (#145)
 
-`0.9.0` (2019-12-20)
+0.9.0 (2019-12-20)
+-------------------
 * Flask support via middleware (#127)
 * Add message attributes to metrics log (#128)
 * Specify number of threads per subscriber with Subscription ThreadPoolExecutor (#139)
 * Publishing timeout while blocking (#137)
 * Clean up rele.config.setup + Worker() init (#132)
 
-`0.8.1` (2019-11-25)
+0.8.1 (2019-11-25)
+-------------------
 * Fix runrele command
 
-`0.8.0` (2019-11-22)
+0.8.0 (2019-11-22)
+-------------------
 * Worker run method (#118)
 * Add kwargs to setup method passed through to middleware (#123)
 * Add missing worker middleware hooks (#121)
 * Add 3.8 support
 * More Documentation
 
-`0.7.0` (2019-10-21)
+0.7.0 (2019-10-21)
+-------------------
 * BREAKING: Remove Django as a dependency (#95)
 * More documentation
 
-`0.6.0` (2019-09-21)
+0.6.0 (2019-09-21)
+-------------------
 * BREAKING: Remove drf as a dependency (#91)
 * Add message as a parameter for middleware hooks (#99)
 * Check setting.CONN_MAX_AGE and warn when not 0 (#97)
 * More documentation
 
-`0.5.0` (2019-08-08)
-* `python manage.py showsubscriptions` command
+0.5.0 (2019-08-08)
+-------------------
+* ``python manage.py showsubscriptions`` command
 * Configurable ENCODER setting
 * Move DEFAULT_ACK_DEADLINE to the RELE config
 * More documentation
 
-`0.4.1` (2019-06-18)
+0.4.1 (2019-06-18)
+-------------------
 * Ability to install app only with rele
 * Define default filter_by in settings.RELE
 
-`0.4.0` (2019-06-17)
+0.4.0 (2019-06-17)
+-------------------
 
-* Set `DEFAULT_ACK_DEADLINE` (#49)
+* Set ``DEFAULT_ACK_DEADLINE`` (#49)
 * Filter by message attributes (#66) 
 * BREAKING: All Relé settings are defined in a dict (#60)
 
 Old structure:
-```python
-from google.oauth2 import service_account
-RELE_GC_CREDENTIALS = service_account.Credentials.from_service_account_file(
-    'rele/settings/dummy-credentials.json'
-)
-RELE_GC_PROJECT_ID = 'dummy-project-id'
-```
 
-New structure:
-```python
-from google.oauth2 import service_account
-RELE = {
-    'GC_CREDENTIALS': service_account.Credentials.from_service_account_file(
+.. code:: python
+
+    from google.oauth2 import service_account
+    RELE_GC_CREDENTIALS = service_account.Credentials.from_service_account_file(
         'rele/settings/dummy-credentials.json'
-    ),
-    'GC_PROJECT_ID': 'dummy-project-id',
-    'MIDDLEWARE': [
-        'rele.contrib.LoggingMiddleware',
-        'rele.contrib.DjangoDBMiddleware',
-    ],
-    'SUB_PREFIX': 'mysubprefix',
-    'APP_NAME': 'myappname',
-}
-```
-* `rele.contrib.middleware` (#55)
+    )
+    RELE_GC_PROJECT_ID = 'dummy-project-id'
+
+New structure:
+
+.. code:: python
+
+    from google.oauth2 import service_account
+    RELE = {
+        'GC_CREDENTIALS': service_account.Credentials.from_service_account_file(
+            'rele/settings/dummy-credentials.json'
+        ),
+        'GC_PROJECT_ID': 'dummy-project-id',
+        'MIDDLEWARE': [
+            'rele.contrib.LoggingMiddleware',
+            'rele.contrib.DjangoDBMiddleware',
+        ],
+        'SUB_PREFIX': 'mysubprefix',
+        'APP_NAME': 'myappname',
+    }
+
+* ``rele.contrib.middleware`` (#55)
 * Prefix argument in sub decorator (#47) 
 * Add timestamp to the published message (#42)
 * BREAKING: Explicit publisher and subscriber configuration (#43)
 * Sphinx documentation (#27, #34, #40, #41)
 * Contributing guidelines (#32)
 
-`0.3.1` (2019-06-04)
+0.3.1 (2019-06-04)
+-------------------
 
 * Add prometheus metrics key to logs (#16 - #20, #22, #23)
 * Fix JSON serialization when publishing (#25)
 
-`0.3.0` (2019-05-14)
+0.3.0 (2019-05-14)
+-------------------
 
 * Ability to run in emulator mode (#12)
 * Add Travis-CI builds (#10)
 * More friendly global publish (#11)
 * Non-blocking behaviour when publishing by default (#6)
 
-`0.2.0` (2019-05-09)
+0.2.0 (2019-05-09)
+-------------------
 
 * Initial version
diff --git a/docs/changelog.rst b/docs/changelog.rst
new file mode 100644
index 0000000..565b052
--- /dev/null
+++ b/docs/changelog.rst
@@ -0,0 +1,1 @@
+.. include:: ../CHANGELOG.rst
diff --git a/docs/index.rst b/docs/index.rst
index e6d62c9..bc9b6b2 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -19,14 +19,15 @@ Release v\ |version|. (`Installation <https://github.com/mercadona/rele>`_)
 **Relé** makes integration with Google PubSub easier and is ready to
 integrate seamlessly into any Django project.
 
-
-Motivation and Features
-_______________________
 The Publish-Subscribe pattern and specifically the Google Cloud Pub/Sub library are
 very powerful tools but you can easily cut your fingers on it. Relé
 makes integration seamless by providing Publisher, Subscriber and Worker
 classes.
 
+
+Features
+________
+
 Out of the box, Relé includes the following features:
 
     * Powerful Publishing API
@@ -84,22 +85,37 @@ ___________
 Configuration
 _____________
 
+Here you can see the full list of the settings options for your deployment of Relé.
+
 .. toctree::
     :maxdepth: 2
 
-    api/settings
+    settings
 
 
 API Docs
 ________
 
 This is the part of documentation that details the inner workings of Relé.
+If you are looking for information on a specific function, class or method,
+this part of the documentation is for you.
 
 
 .. toctree::
     :maxdepth: 2
 
-    api/reference
+    reference
+
+Changelog
+---------
+
+Here you can see the full list of changes between each Relé release.
+
+.. toctree::
+   :maxdepth: 1
+
+   changelog
+
 
 Project Info
 ____________
diff --git a/docs/api/reference.rst b/docs/reference.rst
similarity index 100%
rename from docs/api/reference.rst
rename to docs/reference.rst
diff --git a/docs/api/settings.rst b/docs/settings.rst
similarity index 100%
rename from docs/api/settings.rst
rename to docs/settings.rst
diff --git a/rele/client.py b/rele/client.py
index d1f3e80..cc39b92 100644
--- a/rele/client.py
+++ b/rele/client.py
@@ -7,6 +7,7 @@ from contextlib import suppress
 import google.auth
 from google.api_core import exceptions
 from google.cloud import pubsub_v1
+from google.cloud.pubsub_v1.exceptions import TimeoutError
 
 from rele.middleware import run_middleware_hook
 
@@ -119,7 +120,9 @@ class Publisher:
         else:
             self._client = pubsub_v1.PublisherClient(credentials=credentials)
 
-    def publish(self, topic, data, blocking=False, timeout=None, **attrs):
+    def publish(
+        self, topic, data, blocking=False, timeout=None, raise_exception=True, **attrs
+    ):
         """Publishes message to Google PubSub topic.
 
         Usage::
@@ -148,7 +151,8 @@ class Publisher:
         :param topic: string topic to publish the data.
         :param data: dict with the content of the message.
         :param blocking: boolean
-        :param timeout: float, default None fallsback to :ref:`settings_publisher_timeout`
+        :param timeout: float, default None falls back to :ref:`settings_publisher_timeout`
+        :param raise_exception: boolean. If True, exceptions coming from PubSub will be raised
         :param attrs: additional string parameters to be published.
         :return: `Future <https://googleapis.github.io/google-cloud-python/latest/pubsub/subscriber/api/futures.html>`_  # noqa
         """
@@ -161,6 +165,13 @@ class Publisher:
         if not blocking:
             return future
 
-        future.result(timeout=timeout or self._timeout)
-        run_middleware_hook("post_publish", topic)
+        try:
+            future.result(timeout=timeout or self._timeout)
+        except TimeoutError as e:
+            run_middleware_hook("post_publish_failure", topic, e, data)
+            if raise_exception:
+                raise e
+        else:
+            run_middleware_hook("post_publish", topic)
+
         return future
diff --git a/rele/contrib/logging_middleware.py b/rele/contrib/logging_middleware.py
index e24a675..db59fa0 100644
--- a/rele/contrib/logging_middleware.py
+++ b/rele/contrib/logging_middleware.py
@@ -48,6 +48,20 @@ class LoggingMiddleware(BaseMiddleware):
             },
         )
 
+    def post_publish_failure(self, topic, exception, message):
+        self._logger.exception(
+            f"Exception raised while publishing message "
+            f"for {topic}: {str(exception.__class__.__name__)}",
+            exc_info=True,
+            extra={
+                "metrics": {
+                    "name": "publications",
+                    "data": {"agent": self._app_name, "topic": topic},
+                },
+                "subscription_message": message,
+            },
+        )
+
     def pre_process_message(self, subscription, message):
         self._logger.debug(
             f"Start processing message for {subscription}",
diff --git a/rele/middleware.py b/rele/middleware.py
index 3f95587..bb1a1b0 100644
--- a/rele/middleware.py
+++ b/rele/middleware.py
@@ -47,6 +47,13 @@ class BaseMiddleware:
         :param topic:
         """
 
+    def post_publish_failure(self, topic, exception, message):
+        """Called after publishing fails.
+        :param topic:
+        :param exception:
+        :param message:
+        """
+
     def pre_process_message(self, subscription, message):
         """Called when the Worker receives a message.
         :param subscription:


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index f546190..7706898 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -5,6 +5,7 @@ from unittest.mock import MagicMock, patch
 
 import pytest
 from google.cloud.pubsub_v1 import PublisherClient
+from google.cloud.pubsub_v1.exceptions import TimeoutError
 
 from rele import Publisher
 from rele.client import Subscriber
@@ -36,7 +37,12 @@ def subscriber(project_id, config):
 
 
 @pytest.fixture
-def publisher(config):
+def mock_future():
+    return MagicMock(spec=concurrent.futures.Future)
+
+
+@pytest.fixture
+def publisher(config, mock_future):
     publisher = Publisher(
         gc_project_id=config.gc_project_id,
         credentials=config.credentials,
@@ -44,7 +50,7 @@ def publisher(config):
         timeout=config.publisher_timeout,
     )
     publisher._client = MagicMock(spec=PublisherClient)
-    publisher._client.publish.return_value = concurrent.futures.Future()
+    publisher._client.publish.return_value = mock_future
 
     return publisher
 
@@ -74,3 +80,18 @@ def custom_encoder():
                 return float(obj)
 
     return DecimalEncoder
+
+
+@pytest.fixture
+def mock_publish_timeout():
+    with patch("rele.client.Publisher.publish") as mock:
+        mock.side_effect = TimeoutError()
+        yield mock
+
+
+@pytest.fixture
+def mock_post_publish_failure():
+    with patch(
+        "rele.contrib.logging_middleware.LoggingMiddleware.post_publish_failure"
+    ) as mock:
+        yield mock
diff --git a/tests/test_client.py b/tests/test_client.py
index 8d2edbb..1325a83 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -5,6 +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.exceptions import TimeoutError
 
 
 @pytest.mark.usefixtures("publisher", "time_mock")
@@ -51,7 +52,6 @@ class TestPublisher:
 
         publisher._client.publish.assert_called_with(ANY, b"1.2", published_at=ANY)
 
-    @patch.object(concurrent.futures.Future, "result")
     def test_publishes_data_with_client_timeout_when_blocking(
         self, mock_future, publisher
     ):
@@ -62,9 +62,8 @@ class TestPublisher:
         publisher._client.publish.assert_called_with(
             ANY, b'{"foo": "bar"}', published_at=ANY
         )
-        mock_future.assert_called_once_with(timeout=100)
+        mock_future.result.assert_called_once_with(timeout=100)
 
-    @patch.object(concurrent.futures.Future, "result")
     def test_publishes_data_with_client_timeout_when_blocking_and_timeout_specified(
         self, mock_future, publisher
     ):
@@ -77,7 +76,55 @@ class TestPublisher:
         publisher._client.publish.assert_called_with(
             ANY, b'{"foo": "bar"}', published_at=ANY
         )
-        mock_future.assert_called_once_with(timeout=50)
+        mock_future.result.assert_called_once_with(timeout=50)
+
+    def test_runs_post_publish_failure_hook_when_future_result_raises_timeout(
+        self, mock_future, publisher, mock_post_publish_failure
+    ):
+        message = {"foo": "bar"}
+        exception = TimeoutError()
+        mock_future.result.side_effect = exception
+
+        with pytest.raises(TimeoutError):
+            publisher.publish(
+                topic="order-cancelled", data=message, myattr="hello", blocking=True
+            )
+        mock_post_publish_failure.assert_called_once_with(
+            "order-cancelled", exception, {"foo": "bar"}
+        )
+
+    def test_raises_when_timeout_error_and_raise_exception_is_true(
+        self, publisher, mock_future
+    ):
+        message = {"foo": "bar"}
+        e = TimeoutError()
+        mock_future.result.side_effect = e
+
+        with pytest.raises(TimeoutError):
+            publisher.publish(
+                topic="order-cancelled",
+                data=message,
+                myattr="hello",
+                blocking=True,
+                raise_exception=True,
+            )
+
+    def test_returns_future_when_timeout_error_and_raise_exception_is_false(
+        self, publisher, mock_future
+    ):
+        message = {"foo": "bar"}
+        e = TimeoutError()
+        mock_future.result.side_effect = e
+
+        result = publisher.publish(
+            topic="order-cancelled",
+            data=message,
+            myattr="hello",
+            blocking=True,
+            raise_exception=False,
+        )
+
+        assert result is mock_future
 
 
 class TestSubscriber:
diff --git a/tests/test_middleware.py b/tests/test_middleware.py
index 793f8b9..b13364d 100644
--- a/tests/test_middleware.py
+++ b/tests/test_middleware.py
@@ -22,4 +22,4 @@ class TestMiddleware:
 
         rele.setup(settings, foo="bar")
         assert mock_middleware_setup.called
-        mock_middleware_setup.call_args_list[0][-1] == {"foo": "bar"}
+        assert mock_middleware_setup.call_args_list[0][-1] == {"foo": "bar"}