Simplify exception serialization to avoid issues with complex exceptions

Author: RealOrangeOne

README.md

@@ -201,22 +201,18 @@ assert result.status == ResultStatus.SUCCEEDED
 
 #### Exceptions
 
-If a task raised an exception, its `.exception` will be the exception raised:
+If a task raised an exception, its `.exception_class` will be the exception class raised:
 
 ```python
-assert isinstance(result.exception, ValueError)
+assert 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:
+Note that this is just the type of exception, and contains no other values. The traceback information is reduced to a string that you can print to help debugging:
 
 ```python
 assert isinstance(result.traceback, str)
 ```
 
-The stack frames, `globals()` and `locals()` are not available.
-
-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/migrations/0012_add_separate_exception_fields.py

@@ -0,0 +1,24 @@
+# Generated by Django 5.1.1 on 2024-11-22 16:32
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("django_tasks_database", "0011_rename_complete_status"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="dbtaskresult",
+            name="exception_class_path",
+            field=models.TextField(default="", verbose_name="exception class path"),
+            preserve_default=False,
+        ),
+        migrations.AddField(
+            model_name="dbtaskresult",
+            name="traceback",
+            field=models.TextField(default="", verbose_name="traceback"),
+            preserve_default=False,
+        ),
+    ]

django_tasks/backends/database/migrations/0013_separate_exception_fields.py

@@ -0,0 +1,24 @@
+# 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
+
+
+def separate_exception_fields(
+    apps: StateApps, schema_editor: BaseDatabaseSchemaEditor
+) -> None:
+    DBTaskResult = apps.get_model("django_tasks_database", "DBTaskResult")
+
+    DBTaskResult.objects.using(schema_editor.connection.alias).update(
+        exception_class_path=models.F("exception_data__exc_type"),
+        traceback=models.F("exception_data__exc_traceback"),
+    )
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("django_tasks_database", "0012_add_separate_exception_fields"),
+    ]
+
+    operations = [migrations.RunPython(separate_exception_fields)]

django_tasks/backends/database/migrations/0014_remove_dbtaskresult_exception_data_and_more.py

@@ -0,0 +1,16 @@
+# Generated by Django 5.1.1 on 2024-11-22 16:32
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("django_tasks_database", "0013_separate_exception_fields"),
+    ]
+
+    operations = [
+        migrations.RemoveField(
+            model_name="dbtaskresult",
+            name="exception_data",
+        ),
+    ]

django_tasks/backends/database/models.py

@@ -20,7 +20,7 @@
     ResultStatus,
     Task,
 )
-from django_tasks.utils import exception_to_dict, retry
+from django_tasks.utils import get_exception_traceback, get_module_path, retry
 
 from .utils import normalize_uuid
 
@@ -101,7 +101,9 @@ class DBTaskResult(GenericBase[P, T], models.Model):
     run_after = models.DateTimeField(_("run after"), null=True)
 
     return_value = models.JSONField(_("return value"), default=None, null=True)
-    exception_data = models.JSONField(_("exception data"), default=None, null=True)
+
+    exception_class_path = models.TextField(_("exception class path"))
+    traceback = models.TextField(_("traceback"))
 
     objects = DBTaskResultQuerySet.as_manager()
 
@@ -145,7 +147,12 @@ def task(self) -> Task[P, T]:
     def task_result(self) -> "TaskResult[T]":
         from .backend import TaskResult
 
-        result = TaskResult[T](
+        try:
+            exception_class = import_string(self.exception_class_path)
+        except ImportError:
+            exception_class = None
+
+        return TaskResult[T](
             db_result=self,
             task=self.task,
             id=normalize_uuid(self.id),
@@ -156,13 +163,10 @@ def task_result(self) -> "TaskResult[T]":
             args=self.args_kwargs["args"],
             kwargs=self.args_kwargs["kwargs"],
             backend=self.backend_name,
+            exception_class=exception_class,
+            traceback=self.traceback,
         )
 
-        object.__setattr__(result, "_return_value", self.return_value)
-        object.__setattr__(result, "_exception_data", self.exception_data)
-
-        return result
-
     @retry(backoff_delay=0)
     def claim(self) -> None:
         """
@@ -177,21 +181,31 @@ def set_succeeded(self, return_value: Any) -> None:
         self.status = ResultStatus.SUCCEEDED
         self.finished_at = timezone.now()
         self.return_value = return_value
-        self.exception_data = None
+        self.exception_class_path = ""
+        self.traceback = ""
         self.save(
-            update_fields=["status", "return_value", "finished_at", "exception_data"]
+            update_fields=[
+                "status",
+                "return_value",
+                "finished_at",
+                "exception_class_path",
+                "traceback",
+            ]
         )
 
     @retry()
     def set_failed(self, exc: BaseException) -> None:
         self.status = ResultStatus.FAILED
         self.finished_at = timezone.now()
-        try:
-            self.exception_data = exception_to_dict(exc)
-        except Exception:
-            logger.exception("Task id=%s unable to save exception", self.id)
-            self.exception_data = None
+        self.exception_class_path = get_module_path(type(exc))
+        self.traceback = get_exception_traceback(exc)
         self.return_value = None
         self.save(
-            update_fields=["status", "finished_at", "exception_data", "return_value"]
+            update_fields=[
+                "status",
+                "return_value",
+                "finished_at",
+                "exception_class_path",
+                "traceback",
+            ]
         )

django_tasks/backends/dummy.py

@@ -47,6 +47,8 @@ def enqueue(
             args=args,
             kwargs=kwargs,
             backend=self.alias,
+            exception_class=None,
+            traceback=None,
         )
 
         if self._get_enqueue_on_commit_for_task(task) is not False:

django_tasks/backends/immediate.py

@@ -10,7 +10,7 @@
 
 from django_tasks.signals import task_enqueued, task_finished
 from django_tasks.task import ResultStatus, Task, TaskResult
-from django_tasks.utils import exception_to_dict, get_random_id, json_normalize
+from django_tasks.utils import get_exception_traceback, get_random_id, json_normalize
 
 from .base import BaseTaskBackend
 
@@ -52,10 +52,9 @@ def _execute_task(self, task_result: TaskResult) -> None:
                 raise
 
             object.__setattr__(task_result, "finished_at", timezone.now())
-            try:
-                object.__setattr__(task_result, "_exception_data", exception_to_dict(e))
-            except Exception:
-                logger.exception("Task id=%s unable to save exception", task_result.id)
+
+            object.__setattr__(task_result, "traceback", get_exception_traceback(e))
+            object.__setattr__(task_result, "exception_class", type(e))
 
             object.__setattr__(task_result, "status", ResultStatus.FAILED)
 
@@ -81,6 +80,8 @@ def enqueue(
             args=args,
             kwargs=kwargs,
             backend=self.alias,
+            exception_class=None,
+            traceback=None,
         )
 
         if self._get_enqueue_on_commit_for_task(task) is not False:

django_tasks/task.py

@@ -8,6 +8,7 @@
     Dict,
     Generic,
     Optional,
+    Type,
     TypeVar,
     Union,
     cast,
@@ -22,8 +23,6 @@
 
 from .exceptions import ResultDoesNotExist
 from .utils import (
-    SerializedExceptionDict,
-    exception_from_dict,
     get_module_path,
     json_normalize,
 )
@@ -38,7 +37,8 @@
 DEFAULT_PRIORITY = 0
 
 TASK_REFRESH_ATTRS = {
-    "_exception_data",
+    "exception_class",
+    "traceback",
     "_return_value",
     "finished_at",
     "started_at",
@@ -255,27 +255,13 @@ class TaskResult(Generic[T]):
     backend: str
     """The name of the backend the task will run on"""
 
-    _return_value: Optional[T] = field(init=False, default=None)
-    _exception_data: Optional[SerializedExceptionDict] = field(init=False, default=None)
+    exception_class: Optional[Type[BaseException]]
+    """The exception raised by the task function"""
 
-    @property
-    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
-        )
+    traceback: Optional[str]
+    """The traceback of the exception if the task failed"""
 
-    @property
-    def traceback(self) -> Optional[str]:
-        """
-        Return the string representation of the traceback of the task if it failed
-        """
-        return (
-            cast(SerializedExceptionDict, self._exception_data)["exc_traceback"]
-            if self.status == ResultStatus.FAILED and self._exception_data is not None
-            else None
-        )
+    _return_value: Optional[T] = field(init=False, default=None)
 
     @property
     def return_value(self) -> Optional[T]:

django_tasks/utils.py

@@ -4,25 +4,11 @@
 import time
 from functools import wraps
 from traceback import format_exception
-from typing import Any, Callable, List, TypedDict, TypeVar
+from typing import Any, Callable, TypeVar
 
 from django.utils.crypto import RANDOM_STRING_CHARS
-from django.utils.module_loading import import_string
 from typing_extensions import ParamSpec
 
-
-class SerializedExceptionDict(TypedDict):
-    """Type for the dictionary holding exception informations in task result
-
-    The task result either stores the result of the task, or the serialized exception
-    information required to reconstitute part of the exception for debugging.
-    """
-
-    exc_type: str
-    exc_args: List[Any]
-    exc_traceback: str
-
-
 T = TypeVar("T")
 P = ParamSpec("P")
 
@@ -74,21 +60,8 @@ def get_module_path(val: Any) -> str:
     return f"{val.__module__}.{val.__qualname__}"
 
 
-def exception_to_dict(exc: BaseException) -> SerializedExceptionDict:
-    return {
-        "exc_type": get_module_path(type(exc)),
-        "exc_args": json_normalize(exc.args),
-        "exc_traceback": "".join(format_exception(type(exc), exc, exc.__traceback__)),
-    }
-
-
-def exception_from_dict(exc_data: SerializedExceptionDict) -> 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"])
+def get_exception_traceback(exc: BaseException) -> str:
+    return "".join(format_exception(type(exc), exc, exc.__traceback__))
 
 
 def get_random_id() -> str:

tests/tests/test_database_backend.py

@@ -476,7 +476,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.exception, ValueError)
+        self.assertEqual(result.exception_class, ValueError)
         assert result.traceback  # So that mypy knows the next line is allowed
         self.assertTrue(
             result.traceback.endswith(
@@ -490,9 +490,7 @@ def test_complex_exception(self) -> None:
         result = test_tasks.complex_exception.enqueue()
         self.assertEqual(DBTaskResult.objects.ready().count(), 1)
 
-        with self.assertNumQueries(
-            9 if connection.vendor == "mysql" else 8
-        ), self.assertLogs("django_tasks.backends.database", level="ERROR"):
+        with self.assertNumQueries(9 if connection.vendor == "mysql" else 8):
             self.run_worker()
 
         self.assertEqual(result.status, ResultStatus.NEW)
@@ -504,8 +502,8 @@ 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.exception)
-        self.assertIsNone(result.traceback)
+        self.assertEqual(result.exception_class, ValueError)
+        self.assertIn('ValueError(ValueError("This task failed"))', result.traceback)  # type: ignore[arg-type]
 
         self.assertEqual(DBTaskResult.objects.ready().count(), 0)
 
@@ -1311,7 +1309,7 @@ def test_repeat_ctrl_c(self) -> None:
 
         result.refresh()
         self.assertEqual(result.status, ResultStatus.FAILED)
-        self.assertIsInstance(result.exception, SystemExit)
+        self.assertEqual(result.exception_class, SystemExit)
 
     @skipIf(sys.platform == "win32", "Windows doesn't support SIGKILL")
     def test_kill(self) -> None:
@@ -1349,7 +1347,7 @@ def test_system_exit_task(self) -> None:
 
         result.refresh()
         self.assertEqual(result.status, ResultStatus.FAILED)
-        self.assertIsInstance(result.exception, SystemExit)
+        self.assertEqual(result.exception_class, SystemExit)
 
     def test_keyboard_interrupt_task(self) -> None:
         result = test_tasks.failing_task_keyboard_interrupt.enqueue()
@@ -1361,7 +1359,7 @@ def test_keyboard_interrupt_task(self) -> None:
 
         result.refresh()
         self.assertEqual(result.status, ResultStatus.FAILED)
-        self.assertIsInstance(result.exception, KeyboardInterrupt)
+        self.assertEqual(result.exception_class, KeyboardInterrupt)
 
     def test_multiple_workers(self) -> None:
         results = [test_tasks.sleep_for.enqueue(0.1) for _ in range(10)]