Store task exceptions (#68)

Author: RealOrangeOne

README.md

@@ -153,6 +153,16 @@ result.refresh()
 assert result.status == ResultStatus.COMPLETE
 ```
 
+#### Exceptions
+
+If a task raised an exception, its `.result` will be the exception raised:
+
+```python
+assert isinstance(result.result, ValueError)
+```
+
+As part of the serialization process for exceptions, some information, such as the traceback information, is lost. If the exception could not be serialized, the `.result` is `None`.
+
 ### Backend introspecting
 
 Because `django-tasks` enables support for multiple different backends, those backends may not support all features, and it can be useful to determine this at runtime to ensure the chosen task queue meets the requirements, or to gracefully degrade functionality if it doesn't.

django_tasks/backends/database/management/commands/db_worker.py

@@ -121,7 +121,7 @@ def run_task(self, db_task_result: DBTaskResult) -> None:
                 db_task_result.task_path,
                 ResultStatus.FAILED,
             )
-            db_task_result.set_failed()
+            db_task_result.set_failed(e)
 
             # If the user tried to terminate, let them
             if isinstance(e, KeyboardInterrupt):

django_tasks/backends/database/models.py

@@ -1,3 +1,4 @@
+import logging
 import uuid
 from typing import TYPE_CHECKING, Any, Generic, Optional, TypeVar
 
@@ -9,7 +10,9 @@
 from typing_extensions import ParamSpec
 
 from django_tasks.task import DEFAULT_QUEUE_NAME, ResultStatus, Task
-from django_tasks.utils import retry
+from django_tasks.utils import exception_to_dict, retry
+
+logger = logging.getLogger("django_tasks.backends.database")
 
 T = TypeVar("T")
 P = ParamSpec("P")
@@ -141,7 +144,12 @@ def set_result(self, result: Any) -> None:
         self.save(update_fields=["status", "result", "finished_at"])
 
     @retry()
-    def set_failed(self) -> None:
+    def set_failed(self, exc: BaseException) -> None:
         self.status = ResultStatus.FAILED
         self.finished_at = timezone.now()
-        self.save(update_fields=["status", "finished_at"])
+        try:
+            self.result = 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"])

django_tasks/backends/immediate.py

@@ -1,3 +1,4 @@
+import logging
 from inspect import iscoroutinefunction
 from typing import TypeVar
 from uuid import uuid4
@@ -7,10 +8,13 @@
 from typing_extensions import ParamSpec
 
 from django_tasks.task import ResultStatus, Task, TaskResult
-from django_tasks.utils import json_normalize
+from django_tasks.utils import exception_to_dict, json_normalize
 
 from .base import BaseTaskBackend
 
+logger = logging.getLogger(__name__)
+
+
 T = TypeVar("T")
 P = ParamSpec("P")
 
@@ -29,16 +33,21 @@ def enqueue(
 
         enqueued_at = timezone.now()
         started_at = timezone.now()
+        result_id = str(uuid4())
         try:
             result = json_normalize(calling_task_func(*args, **kwargs))
             status = ResultStatus.COMPLETE
-        except Exception:
-            result = None
+        except Exception as e:
+            try:
+                result = exception_to_dict(e)
+            except Exception:
+                logger.exception("Task id=%s unable to save exception", result_id)
+                result = None
             status = ResultStatus.FAILED
 
         task_result = TaskResult[T](
             task=task,
-            id=str(uuid4()),
+            id=result_id,
             status=status,
             enqueued_at=enqueued_at,
             started_at=started_at,

django_tasks/task.py

@@ -11,6 +11,7 @@
     Optional,
     TypeVar,
     Union,
+    cast,
     overload,
 )
 
@@ -20,6 +21,7 @@
 from typing_extensions import ParamSpec, Self
 
 from .exceptions import ResultDoesNotExist
+from .utils import exception_from_dict, get_module_path
 
 if TYPE_CHECKING:
     from .backends.base import BaseTaskBackend
@@ -147,7 +149,7 @@ def get_backend(self) -> "BaseTaskBackend":
 
     @property
     def module_path(self) -> str:
-        return f"{self.func.__module__}.{self.func.__qualname__}"
+        return get_module_path(self.func)
 
 
 # Bare decorator usage
@@ -220,20 +222,26 @@ class TaskResult(Generic[T]):
     backend: str
     """The name of the backend the task will run on"""
 
-    _result: Optional[T] = field(init=False, default=None)
+    _result: Optional[Union[T, dict]] = field(init=False, default=None)
 
     @property
-    def result(self) -> T:
-        if self.status not in [ResultStatus.COMPLETE, ResultStatus.FAILED]:
-            raise ValueError("Task has not finished yet")
-
-        return self._result  # type:ignore
+    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(dict, self._result))
+                if self._result is not None
+                else None
+            )
+
+        raise ValueError("Task has not finished yet")
 
     def get_result(self) -> Optional[T]:
         """
-        A convenience method to get the result, or None if it's not ready yet.
+        A convenience method to get the result, or None if it's not ready yet or has failed.
         """
-        return self._result
+        return cast(T, self.result) if self.status == ResultStatus.COMPLETE else None
 
     def refresh(self) -> None:
         """

django_tasks/utils.py

@@ -5,6 +5,7 @@
 from functools import wraps
 from typing import Any, Callable, TypeVar
 
+from django.utils.module_loading import import_string
 from typing_extensions import ParamSpec
 
 T = TypeVar("T")
@@ -64,3 +65,23 @@ def inner_wrapper(*args: P.args, **kwargs: P.kwargs) -> T:  # type:ignore[return
         return inner_wrapper
 
     return wrapper
+
+
+def get_module_path(val: Any) -> str:
+    return f"{val.__module__}.{val.__qualname__}"
+
+
+def exception_to_dict(exc: BaseException) -> dict:
+    return {
+        "exc_type": get_module_path(type(exc)),
+        "exc_args": json_normalize(exc.args),
+    }
+
+
+def exception_from_dict(exc_data: dict) -> BaseException:
+    exc_class = import_string(exc_data["exc_type"])
+
+    if not inspect.isclass(exc_class) or not issubclass(exc_class, BaseException):
+        raise TypeError(f"{type(exc_class)} is not an exception")
+
+    return exc_class(*exc_data["exc_args"])

tests/tasks.py

@@ -26,6 +26,11 @@ def failing_task() -> None:
     raise ValueError("This task failed")
 
 
+@task()
+def complex_exception() -> None:
+    raise ValueError(ValueError("This task failed"))
+
+
 @task()
 def exit_task() -> None:
     exit(1)

tests/tests/test_database_backend.py

@@ -44,6 +44,7 @@ def test_enqueue_task(self) -> None:
                 self.assertIsNone(result.finished_at)
                 with self.assertRaisesMessage(ValueError, "Task has not finished yet"):
                     result.result  # noqa:B018
+                self.assertIsNone(result.get_result())
                 self.assertEqual(result.task, task)
                 self.assertEqual(result.args, [1])
                 self.assertEqual(result.kwargs, {"two": 3})
@@ -58,6 +59,7 @@ async def test_enqueue_task_async(self) -> None:
                 self.assertIsNone(result.finished_at)
                 with self.assertRaisesMessage(ValueError, "Task has not finished yet"):
                     result.result  # noqa:B018
+                self.assertIsNone(result.get_result())
                 self.assertEqual(result.task, task)
                 self.assertEqual(result.args, [])
                 self.assertEqual(result.kwargs, {})
@@ -307,6 +309,28 @@ def test_failing_task(self) -> None:
         self.assertGreaterEqual(result.started_at, result.enqueued_at)  # type: ignore
         self.assertGreaterEqual(result.finished_at, result.started_at)  # type: ignore
         self.assertEqual(result.status, ResultStatus.FAILED)
+        self.assertIsInstance(result.result, ValueError)
+
+        self.assertEqual(DBTaskResult.objects.ready().count(), 0)
+
+    def test_complex_exception(self) -> None:
+        result = test_tasks.complex_exception.enqueue()
+        self.assertEqual(DBTaskResult.objects.ready().count(), 1)
+
+        with self.assertNumQueries(8), self.assertLogs(
+            "django_tasks.backends.database", level="ERROR"
+        ):
+            self.run_worker()
+
+        self.assertEqual(result.status, ResultStatus.NEW)
+        result.refresh()
+        self.assertIsNotNone(result.started_at)
+        self.assertIsNotNone(result.finished_at)
+
+        self.assertGreaterEqual(result.started_at, result.enqueued_at)  # type: ignore
+        self.assertGreaterEqual(result.finished_at, result.started_at)  # type: ignore
+        self.assertEqual(result.status, ResultStatus.FAILED)
+        self.assertIsNone(result.result)
 
         self.assertEqual(DBTaskResult.objects.ready().count(), 0)
 

tests/tests/test_immediate_backend.py

@@ -44,6 +44,7 @@ async def test_enqueue_task_async(self) -> None:
                 self.assertGreaterEqual(result.started_at, result.enqueued_at)
                 self.assertGreaterEqual(result.finished_at, result.started_at)
                 self.assertIsNone(result.result)
+                self.assertIsNone(result.get_result())
                 self.assertEqual(result.task, task)
                 self.assertEqual(result.args, [])
                 self.assertEqual(result.kwargs, {})
@@ -56,11 +57,27 @@ def test_catches_exception(self) -> None:
         self.assertIsNotNone(result.finished_at)
         self.assertGreaterEqual(result.started_at, result.enqueued_at)
         self.assertGreaterEqual(result.finished_at, result.started_at)
-        self.assertIsNone(result.result)
+        self.assertIsInstance(result.result, ValueError)
+        self.assertIsNone(result.get_result())
         self.assertEqual(result.task, test_tasks.failing_task)
         self.assertEqual(result.args, [])
         self.assertEqual(result.kwargs, {})
 
+    def test_complex_exception(self) -> None:
+        with self.assertLogs("django_tasks.backends.immediate", level="ERROR"):
+            result = default_task_backend.enqueue(test_tasks.complex_exception, [], {})
+
+        self.assertEqual(result.status, ResultStatus.FAILED)
+        self.assertIsNotNone(result.started_at)
+        self.assertIsNotNone(result.finished_at)
+        self.assertGreaterEqual(result.started_at, result.enqueued_at)
+        self.assertGreaterEqual(result.finished_at, result.started_at)
+        self.assertIsNone(result.result)
+        self.assertIsNone(result.get_result())
+        self.assertEqual(result.task, test_tasks.complex_exception)
+        self.assertEqual(result.args, [])
+        self.assertEqual(result.kwargs, {})
+
     def test_result(self) -> None:
         result = default_task_backend.enqueue(
             test_tasks.calculate_meaning_of_life, [], {}

tests/tests/test_utils.py

@@ -2,9 +2,11 @@
 import subprocess
 from unittest.mock import Mock
 
+from django.core.exceptions import ImproperlyConfigured
 from django.test import SimpleTestCase
 
 from django_tasks import utils
+from django_tasks.exceptions import InvalidTaskError
 from tests import tasks as test_tasks
 
 
@@ -85,3 +87,34 @@ def test_retry(self) -> None:
     def test_keeps_return_value(self) -> None:
         self.assertTrue(utils.retry()(lambda: True)())
         self.assertFalse(utils.retry()(lambda: False)())
+
+
+class ExceptionSerializationTestCase(SimpleTestCase):
+    def test_serialize_exceptions(self) -> None:
+        for exc in [
+            ValueError(10),
+            SyntaxError("Wrong"),
+            ImproperlyConfigured("It's wrong"),
+            InvalidTaskError(""),
+            SystemExit(),
+        ]:
+            with self.subTest(exc):
+                data = utils.exception_to_dict(exc)
+                self.assertEqual(utils.json_normalize(data), data)
+                self.assertEqual(set(data.keys()), {"exc_type", "exc_args"})
+                reconstructed = utils.exception_from_dict(data)
+                self.assertIsInstance(reconstructed, type(exc))
+                self.assertEqual(reconstructed.args, exc.args)
+
+    def test_cannot_deserialize_non_exception(self) -> None:
+        for data in [
+            {"exc_type": "subprocess.check_output", "exc_args": ["exit", "1"]},
+            {"exc_type": "True", "exc_args": []},
+            {"exc_type": "math.pi", "exc_args": []},
+            {"exc_type": __name__, "exc_args": []},
+            {"exc_type": utils.get_module_path(type(self)), "exc_args": []},
+            {"exc_type": utils.get_module_path(Mock), "exc_args": []},
+        ]:
+            with self.subTest(data):
+                with self.assertRaises((TypeError, ImportError)):
+                    utils.exception_from_dict(data)