Simplify exception serialization to avoid issues with complex exceptions (#121)

* Simplify exception serialization to avoid issues with complex exceptions

* Raise exception when accessing exception details on unfinished task

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,30 @@
+# 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.db.models.functions import Coalesce
+
+
+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=Coalesce(
+            models.F("exception_data__exc_type"), models.Value("", models.JSONField())
+        ),
+        traceback=Coalesce(
+            models.F("exception_data__exc_traceback"),
+            models.Value("", models.JSONField()),
+        ),
+    )
+
+
+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.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
+
+        task_result = TaskResult[T](
             db_result=self,
             task=self.task,
             id=normalize_uuid(self.id),
@@ -158,10 +165,10 @@ def task_result(self) -> "TaskResult[T]":
             backend=self.backend_name,
         )
 
-        object.__setattr__(result, "_return_value", self.return_value)
-        object.__setattr__(result, "_exception_data", self.exception_data)
+        object.__setattr__(task_result, "_exception_class", exception_class)
+        object.__setattr__(task_result, "_traceback", self.traceback or None)
 
-        return result
+        return task_result
 
     @retry(backoff_delay=0)
     def claim(self) -> None:
@@ -177,21 +184,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/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)
 

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,10 @@ 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)
-
-    @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
-        )
+    _exception_class: Optional[Type[BaseException]] = field(init=False, default=None)
+    _traceback: Optional[str] = field(init=False, default=None)
 
-    @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]:
@@ -285,14 +268,32 @@ def return_value(self) -> Optional[T]:
         If the task didn't succeed, an exception is raised.
         This is to distinguish against the task returning None.
         """
-        if self.status == ResultStatus.FAILED:
-            raise ValueError("Task failed")
-
-        elif self.status != ResultStatus.SUCCEEDED:
+        if not self.is_finished:
             raise ValueError("Task has not finished yet")
 
         return cast(T, self._return_value)
 
+    @property
+    def exception_class(self) -> Optional[Type[BaseException]]:
+        """The exception raised by the task function"""
+        if not self.is_finished:
+            raise ValueError("Task has not finished yet")
+
+        return self._exception_class
+
+    @property
+    def traceback(self) -> Optional[str]:
+        """The traceback of the exception if the task failed"""
+        if not self.is_finished:
+            raise ValueError("Task has not finished yet")
+
+        return self._traceback
+
+    @property
+    def is_finished(self) -> bool:
+        """Has the task finished?"""
+        return self.status in {ResultStatus.FAILED, ResultStatus.SUCCEEDED}
+
     def refresh(self) -> None:
         """
         Reload the cached task data from the task store

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: