Repository: ansible/awx
File: awx/api/views/__init__.py (Python)

Code before:

class WorkflowJobDetail(UnifiedJobDeletionMixin, RetrieveDestroyAPIView):
    model = models.WorkflowJob
    serializer_class = serializers.WorkflowJobSerializer
    resource_purpose = 'workflow job detail'


class WorkflowJobWorkflowNodesList(SubListAPIView):
    model = models.WorkflowJobNode
    serializer_class = serializers.WorkflowJobNodeListSerializer
    always_allow_superuser = True
    parent_model = models.WorkflowJob
    relationship = 'workflow_job_nodes'
    parent_key = 'workflow_job'
    search_fields = ('unified_job_template__name', 'unified_job_template__description')
    ordering = ('id',)  # assure ordering by id for consistency
    filter_read_permission = False
    resource_purpose = 'workflow nodes of a workflow job'


class WorkflowJobCancel(GenericCancelView):
    model = models.WorkflowJob
    serializer_class = serializers.WorkflowJobCancelSerializer
    resource_purpose = 'cancel for a workflow job'

    @extend_schema_if_available(extensions={"x-ai-description": "Cancel a running workflow job"})
    def post(self, request, *args, **kwargs):
        r = super().post(request, *args, **kwargs)
        ScheduleWorkflowManager().schedule()
        return r


class WorkflowJobNotificationsList(SubListAPIView):
    model = models.Notification
    serializer_class = serializers.NotificationSerializer
    parent_model = models.WorkflowJob
    relationship = 'notifications'
    search_fields = ('subject', 'notification_type', 'body')
    resource_purpose = 'notifications of a workflow job'

    def get_sublist_queryset(self, parent):
        return self.model.objects.filter(
            Q(unifiedjob_notifications=parent)
            | Q(unifiedjob_notifications__unified_job_node__workflow_job=parent, unifiedjob_notifications__workflowapproval__isnull=False)
        ).distinct()


class WorkflowJobActivityStreamList(SubListAPIView):
    model = models.ActivityStream
    serializer_class = serializers.ActivityStreamSerializer
    parent_model = models.WorkflowJob

Review comment (suggestion):
```suggestion
    @extend_schema_if_available(extensions={"x-ai-description": "Cancel a running or pending workflow job"})
```

Code after:
    resource_purpose = 'activity stream of a workflow job template'

    def get_queryset(self):
        parent = self.get_parent_object()
        self.check_parent_access(parent)
        qs = self.request.user.get_queryset(self.model)
        return qs.filter(Q(workflow_job_template=parent) | Q(workflow_job_template_node__workflow_job_template=parent)).distinct()


class WorkflowJobList(ListAPIView):
    model = models.WorkflowJob
    serializer_class = serializers.WorkflowJobListSerializer
    resource_purpose = 'workflow jobs'

    @extend_schema_if_available(extensions={"x-ai-description": "A list of workflow jobs."})
    def get(self, request, *args, **kwargs):
        return super().get(request, *args, **kwargs)


class WorkflowJobDetail(UnifiedJobDeletionMixin, RetrieveDestroyAPIView):
    model = models.WorkflowJob
    serializer_class = serializers.WorkflowJobSerializer
    resource_purpose = 'workflow job detail'


class WorkflowJobWorkflowNodesList(SubListAPIView):
    model = models.WorkflowJobNode
    serializer_class = serializers.WorkflowJobNodeListSerializer
    always_allow_superuser = True
    parent_model = models.WorkflowJob
    relationship = 'workflow_job_nodes'
    parent_key = 'workflow_job'
    search_fields = ('unified_job_template__name', 'unified_job_template__description')
    ordering = ('id',)  # assure ordering by id for consistency
    filter_read_permission = False
    resource_purpose = 'workflow nodes of a workflow job'


class WorkflowJobCancel(GenericCancelView):
    model = models.WorkflowJob
    serializer_class = serializers.WorkflowJobCancelSerializer
    resource_purpose = 'cancel for a workflow job'

    @extend_schema_if_available(extensions={"x-ai-description": "Cancel a running or pending workflow job"})
    def post(self, request, *args, **kwargs):
        r = super().post(request, *args, **kwargs)
        ScheduleWorkflowManager().schedule()
        return r


class WorkflowJobNotificationsList(SubListAPIView):