fix: 终断任务时接口报错

apps/ops/api/job.py

@@ -1,9 +1,11 @@
 import json
 import os
 
+from celery.result import AsyncResult
 from django.conf import settings
 from django.db import transaction
 from django.db.models import Count
+from django.http import Http404
 from django.shortcuts import get_object_or_404
 from django.utils._os import safe_join
 from django.utils.translation import gettext_lazy as _
@@ -14,6 +16,7 @@ from rest_framework.views import APIView
 from assets.models import Asset
 from common.const.http import POST
 from common.permissions import IsValidUser
+from ops.celery import app
 from ops.const import Types
 from ops.models import Job, JobExecution
 from ops.serializers.job import JobSerializer, JobExecutionSerializer, FileSerializer, JobTaskStopSerializer
@@ -194,7 +197,23 @@ class JobExecutionViewSet(OrgBulkModelViewSet):
         if not serializer.is_valid():
             return Response({'error': serializer.errors}, status=400)
         task_id = serializer.validated_data['task_id']
-        instance = get_object_or_404(JobExecution, task_id=task_id, creator=request.user)
+        try:
+            instance = get_object_or_404(JobExecution, task_id=task_id, creator=request.user)
+        except Http404:
+            return Response(
+                {'error': _('The task is being created and cannot be interrupted. Please try again later.')},
+                status=400
+            )
+
+        task = AsyncResult(task_id, app=app)
+        inspect = app.control.inspect()
+        for worker in inspect.registered().keys():
+            if task_id not in [at['id'] for at in inspect.active().get(worker, [])]:
+                # 在队列中未执行使用revoke执行
+                task.revoke(terminate=True)
+                instance.set_error('Job stop by "revoke task {}"'.format(task_id))
+                return Response({'task_id': task_id}, status=200)
+
         instance.stop()
         return Response({'task_id': task_id}, status=200)