Merge pull request #35 from octue/feature/disable-enqueueing-task-using-setting

Disable enqueueing task using setting

Author: thclark

django_gcp/exceptions.py

@@ -36,3 +36,7 @@ class AttemptedOverwriteError(Exception):
 
 class MissingBlobError(Exception):
     """Raised when attempting to access or copy a blob that is not present in GCS"""
+
+
+class IncompatibleSettingsError(ValueError):
+    """Raised when settings values are mutually exclusive"""

django_gcp/tasks/manager.py

@@ -49,6 +49,11 @@ def delimiter(self):
         """Return the GCP_TASKS_DELIMITER setting or a default"""
         return getattr(settings, "GCP_TASKS_DELIMITER", "--")
 
+    @property
+    def disable_execute(self):
+        """Return the GCP_TASKS_DISABLE_EXECUTE setting or default False"""
+        return getattr(settings, "GCP_TASKS_DISABLE_EXECUTE", False)
+
     @property
     def domain(self):
         """Return the GCP_TASKS_DOMAIN setting or a default"""

django_gcp/tasks/tasks.py

@@ -11,7 +11,7 @@
 from django.urls import reverse
 from django.utils.timezone import now
 from django_gcp.events.utils import decode_pubsub_message
-from django_gcp.exceptions import DuplicateTaskError, IncorrectTaskUsageError
+from django_gcp.exceptions import DuplicateTaskError, IncompatibleSettingsError, IncorrectTaskUsageError
 from gcp_pilot.pubsub import CloudPublisher, CloudSubscriber
 from google.api_core.exceptions import AlreadyExists
 from google.cloud import pubsub_v1
@@ -83,13 +83,19 @@ class Task(metaclass=TaskMeta):
     deduplicate = False
 
     def enqueue(self, **kwargs):
-        """Invoke a task (place it onto a queue for processing)"""
+        """Invoke a task (place it onto a queue for processing)
+
+        Some settings affect the behaviour of this method, allowing bypass or immediate execution. See settings documentation for more.
+        """
         return self._send(
             task_kwargs=kwargs,
         )
 
     def enqueue_later(self, when, **kwargs):
-        """Invoke a task (place it onto a queue for processing after some time delay)"""
+        """Invoke a task (place it onto a queue for processing after some time delay)
+
+        Some settings affect the behaviour of this method, allowing bypass or immediate execution. See settings documentation for more.
+        """
         if isinstance(when, int):
             delay_in_seconds = when
         elif isinstance(when, timedelta):
@@ -163,6 +169,12 @@ def _body_to_kwargs(self, request_body):
     # interacting with Cloud Tasks, like the following _send method
 
     def _send(self, task_kwargs, api_kwargs=None):
+        if self.manager.disable_execute and self.manager.eager_execute:
+            raise IncompatibleSettingsError("disable_execute and eager_execute should be mutually exclusive")
+
+        if self.manager.disable_execute:
+            return None
+
         payload = serialize(task_kwargs)
         if self.manager.eager_execute:
             return self.run(**deserialize(payload))

docs/source/tasks_settings.rst

@@ -87,3 +87,18 @@ If set to true, tasks will synchronously execute when their ``enqueue()`` method
 
 Whilst not generally useful in production, this can be quite helpful for straightforward
 debugging of tasks in local environments.
+
+
+``GCP_TASKS_DISABLE_EXECUTE``
+-----------------------------
+Type: ``boolean``
+
+Default: ``False``
+
+If set to true, tasks will not be enqueued for processing when their ``enqueue()`` or ``enqueue_later()``
+methods are called. Instead, the method will simply return None without enqueuing the task, allowing for the
+disabling of task execution. This can be useful in scenarios where tasks need to be temporarily disabled or
+when testing/debugging task code.
+
+It is important to note that this setting only affect tasks when their ``enqueue()`` or ``enqueue_later()``
+methods are called and that tasks can still be executed manually even if this setting is set to True.

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "django-gcp"
-version = "0.9.1"
+version = "0.9.2"
 description = "Utilities to run Django on Google Cloud Platform"
 authors = ["Tom Clark"]
 license = "MIT"

tests/test_tasks_enqueuing.py

@@ -8,15 +8,21 @@
 
 import json
 from unittest.mock import patch
-from django.test import SimpleTestCase
+from django.test import SimpleTestCase, override_settings
 from django.urls import reverse
 from django_gcp.events.utils import make_pubsub_message
-from django_gcp.exceptions import DuplicateTaskError, IncorrectTaskUsageError
+from django_gcp.exceptions import DuplicateTaskError, IncompatibleSettingsError, IncorrectTaskUsageError
 from django_gcp.tasks import OnDemandTask
 from gcp_pilot.mocker import patch_auth
 from google.api_core.exceptions import AlreadyExists
 
-from tests.server.example.tasks import DeduplicatedOnDemandTask, MyOnDemandTask
+from tests.server.example.tasks import (
+    DeduplicatedOnDemandTask,
+    FailingOnDemandTask,
+    MyOnDemandTask,
+    MyPeriodicTask,
+    MySubscriberTask,
+)
 from .test_events_utils import DEFAULT_SUBSCRIPTION
 
 
@@ -79,3 +85,29 @@ def test_failing_on_demand_task_returns_error(self):
 
         self.assertEqual(500, response.status_code)
         self.assertIn("error", response.json())
+
+    def test_disable_enqueueing_with_a_setting(self):
+        """Assert that no task is enqueued if the GCP_TASKS_DISABLE_EXECUTE is true"""
+        with patch_auth():
+            with patch("django_gcp.tasks.tasks.run_coroutine") as patched_send:
+                MyOnDemandTask().enqueue(a="1")
+                self.assertEqual(patched_send.call_count, 1)
+
+                with override_settings(GCP_TASKS_DISABLE_EXECUTE=True):
+                    MyOnDemandTask().enqueue(a="1")
+                    MyPeriodicTask().enqueue(a="1")
+                    MySubscriberTask().enqueue(a="1")
+                    FailingOnDemandTask().enqueue(a="1")
+
+                    self.assertEqual(patched_send.call_count, 1)
+
+                    MyOnDemandTask().enqueue_later(a="1", when=10)
+                    MyPeriodicTask().enqueue_later(a="1", when=10)
+                    MySubscriberTask().enqueue_later(a="1", when=10)
+                    FailingOnDemandTask().enqueue_later(a="1", when=10)
+
+                    self.assertEqual(patched_send.call_count, 1)
+
+            with override_settings(GCP_TASKS_DISABLE_EXECUTE=True, GCP_TASKS_EAGER_EXECUTE=True):
+                with self.assertRaises(IncompatibleSettingsError):
+                    MyOnDemandTask().enqueue(a="1")