Add terminate selected tasks action in admin

Author: caoqianming

django_celery_results/admin.py

@@ -3,6 +3,9 @@
 from django.conf import settings
 from django.contrib import admin
 from django.utils.translation import gettext_lazy as _
+from django.contrib import messages
+from celery import current_app as celery_app
+
 
 try:
     ALLOW_EDITS = settings.DJANGO_CELERY_RESULTS['ALLOW_EDITS']
@@ -58,6 +61,7 @@ class TaskResultAdmin(admin.ModelAdmin):
             'classes': ('extrapretty', 'wide')
         }),
     )
+    actions = ['terminate_task']
 
     def get_readonly_fields(self, request, obj=None):
         if ALLOW_EDITS:
@@ -67,6 +71,26 @@ def get_readonly_fields(self, request, obj=None):
                 field.name for field in self.opts.local_fields
             })
 
+    def terminate_task(self, request, queryset):
+        """Terminate selected tasks."""
+        for task_result in queryset:
+            task_id = task_result.task_id
+            try:
+                celery_app.control.revoke(task_id, terminate=True)
+                self.message_user(
+                    request,
+                    f"Task {task_id} was terminated successfully.",
+                    messages.SUCCESS,
+                )
+            except Exception as e:
+                self.message_user(
+                    request,
+                    f"Failed to terminate task {task_id}. Error: {e}",
+                    messages.ERROR,
+                )
+
+    terminate_task.short_description = "Terminate selected tasks"
+
 
 admin.site.register(TaskResult, TaskResultAdmin)