Add a task_name property to DBTaskResult (#127)

If a task still exists - use that for the name.

However if the underlying function no longer exists - use task_path to come up with an appropriate name to show in the admin.

Author: tomkins

django_tasks/backends/database/admin.py

@@ -10,7 +10,7 @@
 class DBTaskResultAdmin(admin.ModelAdmin):
     list_display = (
         "id",
-        "get_task_name",
+        "task_name",
         "status",
         "enqueued_at",
         "started_at",
@@ -40,7 +40,3 @@ def get_readonly_fields(
         self, request: HttpRequest, obj: Optional[DBTaskResult] = None
     ) -> List[str]:
         return [f.name for f in self.model._meta.fields]
-
-    @admin.display(description="Task")
-    def get_task_name(self, obj: DBTaskResult) -> str:
-        return obj.task.name

django_tasks/backends/database/models.py

@@ -170,6 +170,20 @@ def task_result(self) -> "TaskResult[T]":
 
         return task_result
 
+    @property
+    def task_name(self) -> str:
+        # If the function for an existing task is no longer available, it'll either raise an
+        # ImportError or ModuleNotFoundError (a subclass of ImportError).
+        try:
+            return self.task.name
+        except ImportError:
+            pass
+
+        try:
+            return self.task_path.rsplit(".", 1)[1]
+        except IndexError:
+            return self.task_path
+
     @retry(backoff_delay=0)
     def claim(self) -> None:
         """

tests/tests/test_database_backend.py

@@ -232,6 +232,22 @@ def test_missing_task_path(self) -> None:
         with self.assertRaises(ImportError):
             _ = db_task_result.task
 
+    def test_task_name(self) -> None:
+        for task_path, expected_task_name in [
+            ("tests.tasks.noop_task", "noop_task"),
+            ("tests.tasks.task_not_found", "task_not_found"),
+            ("tests.tasks.module_not_found.module_not_found", "module_not_found"),
+            ("unexpected_function", "unexpected_function"),
+        ]:
+            with self.subTest(task_path):
+                db_task_result = DBTaskResult.objects.create(
+                    args_kwargs={"args": [], "kwargs": {}},
+                    task_path=task_path,
+                    backend_name="default",
+                )
+
+                self.assertEqual(db_task_result.task_name, expected_task_name)
+
     def test_check(self) -> None:
         errors = list(default_task_backend.check())