Separate return value and exception attributes for task results

RealOrangeOne · web-flow · commit 9aa591f1bf0e · 2024-08-30T13:26:51.000+01:00
A partial solution for #19
diff --git a/README.md b/README.md
@@ -182,11 +182,11 @@ default_task_backend.get_result(result_id)
 
 ### Return values
 
-If your task returns something, it can be retrieved from the `.result` attribute on a `TaskResult`. Accessing this property on an unfinished task (ie not `COMPLETE` or `FAILED`) will raise a `ValueError`.
+If your task returns something, it can be retrieved from the `.return_value` attribute on a `TaskResult`. Accessing this property on an unfinished task (ie not `COMPLETE` or `FAILED`) will raise a `ValueError`.
 
 ```python
 assert result.status == ResultStatus.COMPLETE
-assert result.result == 42
+assert result.return_value == 42
 ```
 
 If a result has been updated in the background, you can call `refresh` on it to update its values. Results obtained using `get_result` will always be up-to-date.
@@ -199,10 +199,10 @@ assert result.status == ResultStatus.COMPLETE
 
 #### Exceptions
 
-If a task raised an exception, its `.result` will be the exception raised:
+If a task raised an exception, its `.exception` will be the exception raised:
 
 ```python
-assert isinstance(result.result, ValueError)
+assert isinstance(result.exception, ValueError)
 ```
 
 As part of the serialization process for exceptions, some information is lost. The traceback information is reduced to a string that you can print to help debugging:
diff --git a/django_tasks/backends/database/management/commands/db_worker.py b/django_tasks/backends/database/management/commands/db_worker.py
@@ -130,7 +130,7 @@ def run_task(self, db_task_result: DBTaskResult) -> None:
 
             # Setting the return and success value inside the error handling,
             # So errors setting it (eg JSON encode) can still be recorded
-            db_task_result.set_result(return_value)
+            db_task_result.set_complete(return_value)
             logger.info(
                 "Task id=%s path=%s state=%s",
                 db_task_result.id,
diff --git a/django_tasks/backends/database/migrations/0007_add_separate_results_fields.py b/django_tasks/backends/database/migrations/0007_add_separate_results_fields.py
@@ -0,0 +1,26 @@
+# Generated by Django 4.2.13 on 2024-08-23 14:38
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("django_tasks_database", "0006_alter_dbtaskresult_args_kwargs_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="dbtaskresult",
+            name="exception_data",
+            field=models.JSONField(
+                default=None, null=True, verbose_name="exception data"
+            ),
+        ),
+        migrations.AddField(
+            model_name="dbtaskresult",
+            name="return_value",
+            field=models.JSONField(
+                default=None, null=True, verbose_name="return value"
+            ),
+        ),
+    ]
diff --git a/django_tasks/backends/database/migrations/0008_separate_results_field.py b/django_tasks/backends/database/migrations/0008_separate_results_field.py
@@ -0,0 +1,47 @@
+# Generated by Django 4.2.13 on 2024-08-23 14:38
+
+from django.db import migrations, models
+from django.db.backends.base.schema import BaseDatabaseSchemaEditor
+from django.db.migrations.state import StateApps
+
+from django_tasks import ResultStatus
+
+
+def separate_results_field(
+    apps: StateApps, schema_editor: BaseDatabaseSchemaEditor
+) -> None:
+    DBTaskResult = apps.get_model("django_tasks_database", "DBTaskResult")
+
+    # If a task completed, the result is its return value
+    DBTaskResult.objects.using(schema_editor.connection.alias).filter(
+        status=ResultStatus.COMPLETE
+    ).update(return_value=models.F("result"))
+
+    # If a task failed, the result is the exception data (or nothing)
+    DBTaskResult.objects.using(schema_editor.connection.alias).filter(
+        status=ResultStatus.FAILED
+    ).update(exception_data=models.F("result"))
+
+
+def merge_results_field(
+    apps: StateApps, schema_editor: BaseDatabaseSchemaEditor
+) -> None:
+    DBTaskResult = apps.get_model("django_tasks_database", "DBTaskResult")
+
+    # If a task completed, the result is its return value
+    DBTaskResult.objects.using(schema_editor.connection.alias).filter(
+        status=ResultStatus.COMPLETE
+    ).update(result=models.F("return_value"))
+
+    # If a task failed, the result is the exception data (or nothing)
+    DBTaskResult.objects.using(schema_editor.connection.alias).filter(
+        status=ResultStatus.FAILED
+    ).update(result=models.F("exception_data"))
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("django_tasks_database", "0007_add_separate_results_fields"),
+    ]
+
+    operations = [migrations.RunPython(separate_results_field, merge_results_field)]
diff --git a/django_tasks/backends/database/migrations/0009_remove_results_field.py b/django_tasks/backends/database/migrations/0009_remove_results_field.py
@@ -0,0 +1,16 @@
+# Generated by Django 4.2.13 on 2024-08-23 14:51
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("django_tasks_database", "0008_separate_results_field"),
+    ]
+
+    operations = [
+        migrations.RemoveField(
+            model_name="dbtaskresult",
+            name="result",
+        ),
+    ]
diff --git a/django_tasks/backends/database/models.py b/django_tasks/backends/database/models.py
@@ -99,7 +99,8 @@ class DBTaskResult(GenericBase[P, T], models.Model):
 
     run_after = models.DateTimeField(_("run after"), null=True)
 
-    result = models.JSONField(_("result"), default=None, null=True)
+    return_value = models.JSONField(_("return value"), default=None, null=True)
+    exception_data = models.JSONField(_("exception data"), default=None, null=True)
 
     objects = DBTaskResultQuerySet.as_manager()
 
@@ -147,7 +148,8 @@ def task_result(self) -> "TaskResult[T]":
             backend=self.backend_name,
         )
 
-        result._result = self.result
+        result._return_value = self.return_value
+        result._exception_data = self.exception_data
 
         return result
 
@@ -161,19 +163,25 @@ def claim(self) -> None:
         self.save(update_fields=["status", "started_at"])
 
     @retry()
-    def set_result(self, result: Any) -> None:
+    def set_complete(self, return_value: Any) -> None:
         self.status = ResultStatus.COMPLETE
         self.finished_at = timezone.now()
-        self.result = result
-        self.save(update_fields=["status", "result", "finished_at"])
+        self.return_value = return_value
+        self.exception_data = None
+        self.save(
+            update_fields=["status", "return_value", "finished_at", "exception_data"]
+        )
 
     @retry()
     def set_failed(self, exc: BaseException) -> None:
         self.status = ResultStatus.FAILED
         self.finished_at = timezone.now()
         try:
-            self.result = exception_to_dict(exc)
+            self.exception_data = exception_to_dict(exc)
         except Exception:
             logger.exception("Task id=%s unable to save exception", self.id)
-            self.result = None
-        self.save(update_fields=["status", "finished_at", "result"])
+            self.exception_data = None
+        self.return_value = None
+        self.save(
+            update_fields=["status", "finished_at", "exception_data", "return_value"]
+        )
diff --git a/django_tasks/backends/immediate.py b/django_tasks/backends/immediate.py
@@ -36,16 +36,15 @@ def _execute_task(self, task_result: TaskResult) -> None:
 
         task_result.started_at = timezone.now()
         try:
-            task_result._result = json_normalize(
+            task_result._return_value = json_normalize(
                 calling_task_func(*task_result.args, **task_result.kwargs)
             )
         except BaseException as e:
             task_result.finished_at = timezone.now()
             try:
-                task_result._result = exception_to_dict(e)
+                task_result._exception_data = exception_to_dict(e)
             except Exception:
                 logger.exception("Task id=%s unable to save exception", task_result.id)
-                task_result._result = None
 
             # Use `.exception` to integrate with error monitoring tools (eg Sentry)
             logger.exception(
diff --git a/django_tasks/task.py b/django_tasks/task.py
@@ -33,6 +33,14 @@
 MAX_PRIORITY = 100
 DEFAULT_PRIORITY = 0
 
+TASK_REFRESH_ATTRS = {
+    "_exception_data",
+    "_return_value",
+    "finished_at",
+    "started_at",
+    "status",
+}
+
 
 class ResultStatus(TextChoices):
     NEW = ("NEW", _("New"))
@@ -239,59 +247,58 @@ class TaskResult(Generic[T]):
     backend: str
     """The name of the backend the task will run on"""
 
-    _result: Optional[Union[T, SerializedExceptionDict]] = field(
-        init=False, default=None
-    )
+    _return_value: Optional[T] = field(init=False, default=None)
+    _exception_data: Optional[SerializedExceptionDict] = field(init=False, default=None)
 
     @property
-    def result(self) -> Optional[Union[T, BaseException]]:
-        if self.status == ResultStatus.COMPLETE:
-            return cast(T, self._result)
-        elif self.status == ResultStatus.FAILED:
-            return (
-                exception_from_dict(cast(SerializedExceptionDict, self._result))
-                if self._result is not None
-                else None
-            )
-
-        raise ValueError("Task has not finished yet")
+    def exception(self) -> Optional[BaseException]:
+        return (
+            exception_from_dict(cast(SerializedExceptionDict, self._exception_data))
+            if self.status == ResultStatus.FAILED and self._exception_data is not None
+            else None
+        )
 
     @property
     def traceback(self) -> Optional[str]:
         """
         Return the string representation of the traceback of the task if it failed
         """
-        if self.status == ResultStatus.FAILED and self._result is not None:
-            return cast(SerializedExceptionDict, self._result)["exc_traceback"]
-
-        return None
+        return (
+            cast(SerializedExceptionDict, self._exception_data)["exc_traceback"]
+            if self.status == ResultStatus.FAILED and self._exception_data is not None
+            else None
+        )
 
-    def get_result(self) -> Optional[T]:
+    @property
+    def return_value(self) -> Optional[T]:
         """
-        A convenience method to get the result, or None if it's not ready yet or has failed.
+        Get the return value of the task.
+
+        If the task didn't complete successfully, an exception is raised.
+        This is to distinguish against the task returning None.
         """
-        return cast(T, self.result) if self.status == ResultStatus.COMPLETE else None
+        if self.status == ResultStatus.FAILED:
+            raise ValueError("Task failed")
+
+        elif self.status != ResultStatus.COMPLETE:
+            raise ValueError("Task has not finished yet")
+
+        return cast(T, self._return_value)
 
     def refresh(self) -> None:
         """
         Reload the cached task data from the task store
         """
         refreshed_task = self.task.get_backend().get_result(self.id)
 
-        # status, started_at, finished_at and result are the only refreshable attributes
-        self.status = refreshed_task.status
-        self.started_at = refreshed_task.started_at
-        self.finished_at = refreshed_task.finished_at
-        self._result = refreshed_task._result
+        for attr in TASK_REFRESH_ATTRS:
+            setattr(self, attr, getattr(refreshed_task, attr))
 
     async def arefresh(self) -> None:
         """
         Reload the cached task data from the task store
         """
         refreshed_task = await self.task.get_backend().aget_result(self.id)
 
-        # status, started_at, finished_at and result are the only refreshable attributes
-        self.status = refreshed_task.status
-        self.started_at = refreshed_task.started_at
-        self.finished_at = refreshed_task.finished_at
-        self._result = refreshed_task._result
+        for attr in TASK_REFRESH_ATTRS:
+            setattr(self, attr, getattr(refreshed_task, attr))
diff --git a/tests/tests/test_database_backend.py b/tests/tests/test_database_backend.py
@@ -69,8 +69,7 @@ def test_enqueue_task(self) -> None:
                 self.assertIsNone(result.started_at)
                 self.assertIsNone(result.finished_at)
                 with self.assertRaisesMessage(ValueError, "Task has not finished yet"):
-                    result.result  # noqa:B018
-                self.assertIsNone(result.get_result())
+                    result.return_value  # noqa:B018
                 self.assertEqual(result.task, task)
                 self.assertEqual(result.args, [1])
                 self.assertEqual(result.kwargs, {"two": 3})
@@ -84,8 +83,7 @@ async def test_enqueue_task_async(self) -> None:
                 self.assertIsNone(result.started_at)
                 self.assertIsNone(result.finished_at)
                 with self.assertRaisesMessage(ValueError, "Task has not finished yet"):
-                    result.result  # noqa:B018
-                self.assertIsNone(result.get_result())
+                    result.return_value  # noqa:B018
                 self.assertEqual(result.task, task)
                 self.assertEqual(result.args, [])
                 self.assertEqual(result.kwargs, {})
@@ -438,7 +436,7 @@ def test_failing_task(self) -> None:
         self.assertGreaterEqual(result.finished_at, result.started_at)  # type: ignore
         self.assertEqual(result.status, ResultStatus.FAILED)
 
-        self.assertIsInstance(result.result, ValueError)
+        self.assertIsInstance(result.exception, ValueError)
         assert result.traceback  # So that mypy knows the next line is allowed
         self.assertTrue(
             result.traceback.endswith(
@@ -466,7 +464,7 @@ def test_complex_exception(self) -> None:
         self.assertGreaterEqual(result.finished_at, result.started_at)  # type: ignore
         self.assertEqual(result.status, ResultStatus.FAILED)
 
-        self.assertIsNone(result.result)
+        self.assertIsNone(result.exception)
         self.assertIsNone(result.traceback)
 
         self.assertEqual(DBTaskResult.objects.ready().count(), 0)
diff --git a/tests/tests/test_dummy_backend.py b/tests/tests/test_dummy_backend.py
@@ -31,7 +31,7 @@ def test_enqueue_task(self) -> None:
                 self.assertIsNone(result.started_at)
                 self.assertIsNone(result.finished_at)
                 with self.assertRaisesMessage(ValueError, "Task has not finished yet"):
-                    result.result  # noqa:B018
+                    result.return_value  # noqa:B018
                 self.assertEqual(result.task, task)
                 self.assertEqual(result.args, [1])
                 self.assertEqual(result.kwargs, {"two": 3})
@@ -47,7 +47,7 @@ async def test_enqueue_task_async(self) -> None:
                 self.assertIsNone(result.started_at)
                 self.assertIsNone(result.finished_at)
                 with self.assertRaisesMessage(ValueError, "Task has not finished yet"):
-                    result.result  # noqa:B018
+                    result.return_value  # noqa:B018
                 self.assertEqual(result.task, task)
                 self.assertEqual(result.args, [])
                 self.assertEqual(result.kwargs, {})
diff --git a/tests/tests/test_immediate_backend.py b/tests/tests/test_immediate_backend.py
diff --git a/tests/views.py b/tests/views.py
