Merge pull request #103 from octue/allow-execute-wrap

Allow wrapping of execute() by subclasses

Author: thclark

django_gcp/tasks/tasks.py

@@ -111,19 +111,15 @@ def enqueue_later(self, when, **kwargs):
             api_kwargs=dict(delay_in_seconds=delay_in_seconds),
         )
 
-    def execute(self, request_body):
-        """Deserialises the received request and calls the run() method"""
-        try:
-            task_kwargs = self._body_to_kwargs(request_body=request_body)
-        except Exception as e:
-            logger.warning(e, exc_info=True)
-            return f"Unable to parse request arguments. Error was: {e}", 400
-
-        try:
-            return self.run(**task_kwargs), 200
-        except Exception as e:
-            logger.error(e, exc_info=True)
-            return "Error running task", 500
+    def execute(self, **task_kwargs):
+        """Executes the run() method
+
+        This simple wrapper allows subsubclasses to retain a simple run()
+        method api, whilst a subclass overloads execute() to add common
+        functionality.
+        """
+
+        return self.run(**task_kwargs)
 
     @property
     def manager(self):
@@ -176,10 +172,10 @@ def _send(self, task_kwargs, api_kwargs=None):
         if self.manager.disable_execute:
             return None
 
-        payload = serialize(task_kwargs)
         if self.manager.eager_execute:
-            return self.execute(payload)
+            return self.execute(**task_kwargs)
 
+        payload = serialize(task_kwargs)
         api_kwargs = api_kwargs or {}
         api_kwargs.update(
             dict(

django_gcp/tasks/views.py

@@ -1,4 +1,5 @@
 import json
+import logging
 from typing import Any, Dict
 
 from django.apps import apps
@@ -7,6 +8,8 @@
 from django.views.decorators.csrf import csrf_exempt
 from django.views.generic import View
 
+logger = logging.getLogger(__name__)
+
 
 @method_decorator(csrf_exempt, name="dispatch")
 class GoogleCloudTaskView(View):
@@ -31,13 +34,22 @@ def post(self, request, task_name, *args, **kwargs):
             result = {"error": f"Task {task_name} not found", "available_tasks": list(self.tasks)}
             return self._prepare_response(status=status, payload=result)
 
-        output, status = task_class().execute(request_body=request.body)
-        if status == 200:
-            result = {"result": output}
-        else:
-            result = {"error": output}
+        task = task_class()
+        try:
+            task_kwargs = task._body_to_kwargs(request_body=request.body)
+        except Exception as e:
+            logger.warning(e, exc_info=True)
+            return self._prepare_response(
+                status=400, payload={"error": f"Unable to parse request arguments. Error was: {e}"}
+            )
+
+        try:
+            result = task.execute(**task_kwargs)
+        except Exception as e:
+            logger.error(e, exc_info=True)
+            return self._prepare_response(status=500, payload={"error": f"Error running task. Error was: {e}"})
 
-        return self._prepare_response(status=status, payload=result)
+        return self._prepare_response(status=200, payload={"result": result})
 
     def _prepare_response(self, status: int, payload: Dict[str, Any]):
         return HttpResponse(status=status, content=json.dumps(payload), content_type="application/json")

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "django-gcp"
-version = "0.23.0"
+version = "0.24.0"
 description = "Utilities to run Django on Google Cloud Platform"
 authors = [{name="Tom Clark"}]
 license = "MIT"

tests/test_tasks_enqueuing.py

@@ -104,3 +104,13 @@ def test_disable_enqueueing_with_a_setting(self):
             with override_settings(GCP_TASKS_DISABLE_EXECUTE=True, GCP_TASKS_EAGER_EXECUTE=True):
                 with self.assertRaises(IncompatibleSettingsError):
                     MyOnDemandTask().enqueue(a="1")
+
+    def test_enqueueing_with_eager_execute(self):
+        """Assert that tasks are successfully executed if GCP_TASKS_EAGER_EXECUTE is true"""
+
+        with patch("tests.server.example.tasks.MyOnDemandTask.run", return_value=None) as patched_run:
+            with override_settings(GCP_TASKS_DISABLE_EXECUTE=False, GCP_TASKS_EAGER_EXECUTE=True):
+                result = MyOnDemandTask().enqueue(a="1")
+
+        self.assertIsNone(result)
+        patched_run.assert_called_once()