Delete and retry tasks through Django Admin

Author: mgax

django_tasks/backends/database/admin.py

@@ -3,9 +3,23 @@
 from django.contrib import admin
 from django.http import HttpRequest
 
+from django_tasks.task import ResultStatus
+
 from .models import DBTaskResult
 
 
+def reenqueue(modeladmin: admin.ModelAdmin, request, queryset):
+    tasks = queryset.update(status=ResultStatus.NEW)
+    modeladmin.message_user(request, f"Rescheduled {tasks} tasks.", "SUCCESS")
+
+
+def duplicate(modeladmin: admin.ModelAdmin, request, queryset):
+    tasks = DBTaskResult.objects.bulk_create(
+        old_task.duplicate() for old_task in queryset
+    )
+    modeladmin.message_user(request, f"Rescheduled {tasks} tasks.", "SUCCESS")
+
+
 @admin.register(DBTaskResult)
 class DBTaskResultAdmin(admin.ModelAdmin):
     list_display = (
@@ -19,17 +33,13 @@ class DBTaskResultAdmin(admin.ModelAdmin):
         "queue_name",
     )
     list_filter = ("status", "priority", "queue_name")
+    actions = [reenqueue, duplicate]
 
     def has_add_permission(
         self, request: HttpRequest, obj: Optional[DBTaskResult] = None
     ) -> bool:
         return False
 
-    def has_delete_permission(
-        self, request: HttpRequest, obj: Optional[DBTaskResult] = None
-    ) -> bool:
-        return False
-
     def has_change_permission(
         self, request: HttpRequest, obj: Optional[DBTaskResult] = None
     ) -> bool:

django_tasks/backends/database/models.py

@@ -169,3 +169,13 @@ def set_failed(self, exc: BaseException) -> None:
             logger.exception("Task id=%s unable to save exception", self.id)
             self.result = None
         self.save(update_fields=["status", "finished_at", "result"])
+
+    def duplicate(self):
+        return type(self)(
+            args_kwargs=self.args_kwargs,
+            priority=self.priority,
+            task_path=self.task_path,
+            queue_name=self.queue_name,
+            backend_name=self.backend_name,
+            run_after=self.run_after,
+        )

tests/tests/test_database_backend.py

@@ -18,6 +18,7 @@
 
 from django_tasks import ResultStatus, Task, default_task_backend, tasks
 from django_tasks.backends.database import DatabaseBackend
+from django_tasks.backends.database.backend import TaskResult
 from django_tasks.backends.database.management.commands.db_worker import (
     logger as db_worker_logger,
 )
@@ -593,6 +594,8 @@ def test_worker_with_locked_rows(self) -> None:
     }
 )
 class DatabaseTaskResultTestCase(TransactionTestCase):
+    run_worker = partial(call_command, "db_worker", verbosity=0, batch=True, interval=0)
+
     def execute_in_new_connection(self, sql: Union[str, QuerySet]) -> Sequence:
         if isinstance(sql, QuerySet):
             sql = str(sql.query)
@@ -792,3 +795,19 @@ def test_get_locked_with_locked_rows(self) -> None:
                 )
         finally:
             new_connection.close()
+
+    def test_duplicate(self) -> None:
+        result_1 = cast(TaskResult, test_tasks.calculate_meaning_of_life.enqueue())
+        db_result_1 = result_1.db_result
+        db_result_2 = result_1.db_result.duplicate()
+        db_result_2.save()
+        result_2 = db_result_2.task_result
+
+        assert db_result_1.pk != db_result_2.pk
+
+        self.run_worker()
+
+        result_1.refresh()
+        result_2.refresh()
+
+        assert result_1.result == result_2.result