stopping_task_pr_1

backend/tasks/urls.py

@@ -6,6 +6,9 @@
     AnnotationViewSet,
     PredictionViewSet,
     get_celery_tasks,
+    stopping_celery_tasks,
+    resume_celery_task,
+    delete_celery_task,
 )
 
 router = routers.DefaultRouter()
@@ -15,4 +18,7 @@
 
 urlpatterns = [
     path("get_celery_tasks/", get_celery_tasks),
+    path("stopping_celery_tasks/", stopping_celery_tasks),
+    path("resume_celery_task/", resume_celery_task),
+    path("delete_celery_task/", delete_celery_task),
 ] + router.urls

backend/tasks/views.py

@@ -11,6 +11,11 @@
 from django.utils import timezone
 from django.core.paginator import Paginator, EmptyPage, PageNotAnInteger
 import json
+from celery import Celery
+
+# from flower.api import Flower
+# flower_app = Flower()
+celery_app = Celery()
 
 from django.core.exceptions import ObjectDoesNotExist
 from django.http import StreamingHttpResponse, FileResponse
@@ -2589,3 +2594,61 @@ def get_celery_tasks(request):
     page_size = int(request.GET.get("page_size", 10))
     data = paginate_queryset(filtered_tasks, page_number, page_size)
     return JsonResponse(data["results"], safe=False)
+
+
+@api_view(["GET"])
+def stopping_celery_tasks(req):
+    task_id = req.GET.get("task_id")
+
+    if task_id is None:
+        return JsonResponse({"message": "Task ID is required"}, status=400)
+
+    task = celery_app.AsyncResult(task_id)
+
+    if task is None or task.state == "PENDING":
+        return JsonResponse({"message": "Task not found or not running"}, status=404)
+
+    if task.state in ["SUCCESS", "FAILURE", "REVOKED"]:
+        return JsonResponse(
+            {"message": "Task already completed or revoked"}, status=400
+        )
+
+    task.revoke(terminate=True)
+
+    return JsonResponse({"message": "Task stopped successfully"}, status=200)
+
+
+@api_view(["GET"])
+def resume_celery_task(req):
+    task_id = req.GET.get("task_id")
+
+    if task_id is None:
+        return JsonResponse({"message": "Task ID is required"}, status=400)
+
+    task = celery_app.AsyncResult(task_id)
+
+    if task is None or task.state not in ["REVOKED", "FAILURE"]:
+        return JsonResponse(
+            {"message": "Task not found or cannot be resumed"}, status=400
+        )
+
+    task.revive()
+
+    return JsonResponse({"message": "Task resumed successfully"}, status=200)
+
+
+@api_view(["GET"])
+def delete_celery_task(req):
+    task_id = req.GET.get("task_id")
+
+    if task_id is None:
+        return JsonResponse({"message": "Task ID is required"}, status=400)
+
+    task = celery_app.AsyncResult(task_id)
+
+    if task is None:
+        return JsonResponse({"message": "Task not found"}, status=404)
+
+    task.forget()
+
+    return JsonResponse({"message": "Task deleted successfully"}, status=200)