<filename>nautobot/extras/api/views.py
from datetime import datetime
from django.contrib.contenttypes.models import ContentType
from django.forms import ValidationError as FormsValidationError
from django.http import Http404
from django.shortcuts import get_object_or_404
from django.utils import timezone
from drf_yasg import openapi
from drf_yasg.utils import no_body, swagger_auto_schema
from graphene_django.views import GraphQLView
from graphql import GraphQLError
from rest_framework import status
from rest_framework.decorators import action
from rest_framework.exceptions import PermissionDenied
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.routers import APIRootView
from rest_framework import viewsets

from nautobot.core.api.metadata import ContentTypeMetadata, StatusFieldMetadata
from nautobot.core.api.views import ModelViewSet, ReadOnlyModelViewSet
from nautobot.core.graphql import execute_saved_query
from nautobot.extras import filters
from nautobot.extras.choices import JobExecutionType, JobResultStatusChoices
from nautobot.extras.datasources import enqueue_pull_git_repository_and_refresh_data
from nautobot.extras.models import (
    ComputedField,
    ConfigContext,
    ConfigContextSchema,
    CustomLink,
    ExportTemplate,
    GitRepository,
    GraphQLQuery,
    ImageAttachment,
    JobLogEntry,
    JobResult,
    ObjectChange,
    Relationship,
    RelationshipAssociation,
    ScheduledJob,
    Secret,
    SecretsGroup,
    SecretsGroupAssociation,
    Status,
    Tag,
    TaggedItem,
    Webhook,
)
from nautobot.extras.models import CustomField, CustomFieldChoice
from nautobot.extras.jobs import get_job, get_jobs, run_job
from nautobot.extras.utils import get_worker_count
from nautobot.utilities.exceptions import CeleryWorkerNotRunningException
from nautobot.utilities.utils import copy_safe_request, count_related
from . import nested_serializers, serializers


class ExtrasRootView(APIRootView):
    """
    Extras API root view
    """

    def get_view_name(self):
        return "Extras"


#
#  Computed Fields
#


class ComputedFieldViewSet(ModelViewSet):
    """
    Manage Computed Fields through DELETE, GET, POST, PUT, and PATCH requests.
    """

    queryset = ComputedField.objects.all()
    serializer_class = serializers.ComputedFieldSerializer
    filterset_class = filters.ComputedFieldFilterSet


#
# Config contexts
#


class ConfigContextQuerySetMixin:
    """
    Used by views that work with config context models (device and virtual machine).
    Provides a get_queryset() method which deals with adding the config context
    data annotation or not.
    """

    def get_queryset(self):
        """
        Build the proper queryset based on the request context

        If the `brief` query param equates to True or the `exclude` query param
        includes `config_context` as a value, return the base queryset.

        Else, return the queryset annotated with config context data
        """
        queryset = super().get_queryset()
        request = self.get_serializer_context()["request"]
        if self.brief or "config_context" in request.query_params.get("exclude", []):
            return queryset
        return queryset.annotate_config_context_data()


class ConfigContextViewSet(ModelViewSet):
    queryset = ConfigContext.objects.prefetch_related(
        "regions",
        "sites",
        "roles",
        "device_types",
        "platforms",
        "tenant_groups",
        "tenants",
    )
    serializer_class = serializers.ConfigContextSerializer
    filterset_class = filters.ConfigContextFilterSet


#
# Config context schemas
#


class ConfigContextSchemaViewSet(ModelViewSet):
    queryset = ConfigContextSchema.objects.all()
    serializer_class = serializers.ConfigContextSchemaSerializer
    filterset_class = filters.ConfigContextSchemaFilterSet


#
# ContentTypes
#


class ContentTypeViewSet(viewsets.ReadOnlyModelViewSet):
    """
    Read-only list of ContentTypes. Limit results to ContentTypes pertinent to Nautobot objects.
    """

    queryset = ContentType.objects.order_by("app_label", "model")
    serializer_class = serializers.ContentTypeSerializer
    filterset_class = filters.ContentTypeFilterSet


#
# Custom fields
#


class CustomFieldViewSet(ModelViewSet):
    metadata_class = ContentTypeMetadata
    queryset = CustomField.objects.all()
    serializer_class = serializers.CustomFieldSerializer
    filterset_class = filters.CustomFieldFilterSet


class CustomFieldChoiceViewSet(ModelViewSet):
    queryset = CustomFieldChoice.objects.all()
    serializer_class = serializers.CustomFieldChoiceSerializer
    filterset_class = filters.CustomFieldChoiceFilterSet


class CustomFieldModelViewSet(ModelViewSet):
    """
    Include the applicable set of CustomFields in the ModelViewSet context.
    """

    def get_serializer_context(self):

        # Gather all custom fields for the model
        content_type = ContentType.objects.get_for_model(self.queryset.model)
        custom_fields = content_type.custom_fields.all()

        context = super().get_serializer_context()
        context.update(
            {
                "custom_fields": custom_fields,
            }
        )
        return context


#
# Custom Links
#


class CustomLinkViewSet(ModelViewSet):
    """
    Manage Custom Links through DELETE, GET, POST, PUT, and PATCH requests.
    """

    queryset = CustomLink.objects.all()
    serializer_class = serializers.CustomLinkSerializer
    filterset_class = filters.CustomLinkFilterSet


#
# Export templates
#


class ExportTemplateViewSet(ModelViewSet):
    metadata_class = ContentTypeMetadata
    queryset = ExportTemplate.objects.all()
    serializer_class = serializers.ExportTemplateSerializer
    filterset_class = filters.ExportTemplateFilterSet


#
# Git repositories
#


class GitRepositoryViewSet(CustomFieldModelViewSet):
    """
    Manage the use of Git repositories as external data sources.
    """

    queryset = GitRepository.objects.all()
    serializer_class = serializers.GitRepositorySerializer
    filterset_class = filters.GitRepositoryFilterSet

    @swagger_auto_schema(method="post", request_body=serializers.GitRepositorySerializer)
    @action(detail=True, methods=["post"])
    def sync(self, request, pk):
        """
        Enqueue pull git repository and refresh data.
        """
        if not request.user.has_perm("extras.change_gitrepository"):
            raise PermissionDenied("This user does not have permission to make changes to Git repositories.")

        if not get_worker_count():
            raise CeleryWorkerNotRunningException()

        repository = get_object_or_404(GitRepository, id=pk)
        enqueue_pull_git_repository_and_refresh_data(repository, request)
        return Response({"message": f"Repository {repository} sync job added to queue."})


#
# GraphQL Queries
#


class GraphQLQueryViewSet(ModelViewSet):
    queryset = GraphQLQuery.objects.all()
    serializer_class = serializers.GraphQLQuerySerializer
    filterset_class = filters.GraphQLQueryFilterSet

    @swagger_auto_schema(
        method="post",
        request_body=serializers.GraphQLQueryInputSerializer,
        responses={"200": serializers.GraphQLQueryOutputSerializer},
    )
    @action(detail=True, methods=["post"])
    def run(self, request, pk):
        try:
            query = get_object_or_404(self.queryset, pk=pk)
            result = execute_saved_query(query.slug, variables=request.data.get("variables"), request=request).to_dict()
            return Response(result)
        except GraphQLError as error:
            return Response(
                {"errors": [GraphQLView.format_error(error)]},
                status=status.HTTP_400_BAD_REQUEST,
            )


#
# Image attachments
#


class ImageAttachmentViewSet(ModelViewSet):
    metadata_class = ContentTypeMetadata
    queryset = ImageAttachment.objects.all()
    serializer_class = serializers.ImageAttachmentSerializer
    filterset_class = filters.ImageAttachmentFilterSet


#
# Jobs
#


class JobViewSet(viewsets.ViewSet):
    permission_classes = [IsAuthenticated]
    lookup_field = "class_path"
    lookup_value_regex = "[^/]+/[^/]+/[^/]+"  # e.g. "git.repo_name/module_name/JobName"

    def _get_job_class(self, class_path):
        job_class = get_job(class_path)
        if job_class is None:
            raise Http404

        return job_class

    def _create_schedule(self, serializer, data, commit, job, job_class, request):
        """
        This is an internal function to create a scheduled job from API data.
        It has to handle boths once-offs (i.e. of type TYPE_FUTURE) and interval
        jobs.
        """
        job_kwargs = {
            "data": data,
            "request": copy_safe_request(request),
            "user": request.user.pk,
            "commit": commit,
            "name": job.class_path,
        }
        type_ = serializer["interval"]
        if type_ == JobExecutionType.TYPE_IMMEDIATELY:
            time = datetime.now()
            name = serializer.get("name") or f"{job.name} - {time}"
        else:
            time = serializer["start_time"]
            name = serializer["name"]
        scheduled_job = ScheduledJob(
            name=name,
            task="nautobot.extras.jobs.scheduled_job_handler",
            job_class=job.class_path,
            start_time=time,
            description=f"Nautobot job {name} scheduled by {request.user} on {time}",
            kwargs=job_kwargs,
            interval=type_,
            one_off=(type_ == JobExecutionType.TYPE_FUTURE),
            user=request.user,
            approval_required=job_class.approval_required,
        )
        scheduled_job.save()
        return scheduled_job

    def list(self, request):
        if not request.user.has_perm("extras.view_job"):
            raise PermissionDenied("This user does not have permission to view jobs.")
        job_content_type = ContentType.objects.get(app_label="extras", model="job")
        results = {
            r.name: r
            for r in JobResult.objects.filter(
                obj_type=job_content_type,
                status__in=JobResultStatusChoices.TERMINAL_STATE_CHOICES,
            )
            .defer("data")
            .order_by("created")
        }

        jobs = get_jobs()
        jobs_list = []
        for grouping, modules in jobs.items():
            for module_name, entry in modules.items():
                for job_class in entry["jobs"].values():
                    job = job_class()
                    job.result = results.get(job.class_path, None)
                    jobs_list.append(job)

        serializer = serializers.JobSerializer(jobs_list, many=True, context={"request": request})

        return Response(serializer.data)

    def retrieve(self, request, class_path):
        if not request.user.has_perm("extras.view_job"):
            raise PermissionDenied("This user does not have permission to view jobs.")
        job_class = self._get_job_class(class_path)
        job_content_type = ContentType.objects.get(app_label="extras", model="job")
        job = job_class()
        job.result = JobResult.objects.filter(
            obj_type=job_content_type,
            name=job.class_path,
            status__in=JobResultStatusChoices.TERMINAL_STATE_CHOICES,
        ).first()

        serializer = serializers.JobDetailSerializer(job, context={"request": request})

        return Response(serializer.data)

    @swagger_auto_schema(method="post", request_body=serializers.JobInputSerializer)
    @action(detail=True, methods=["post"])
    def run(self, request, class_path):
        if not request.user.has_perm("extras.run_job"):
            raise PermissionDenied("This user does not have permission to run jobs.")

        job_class = self._get_job_class(class_path)
        job = job_class()

        input_serializer = serializers.JobInputSerializer(data=request.data)
        input_serializer.is_valid(raise_exception=True)

        data = input_serializer.data["data"] or {}
        commit = input_serializer.data["commit"]
        if commit is None:
            commit = getattr(job_class.Meta, "commit_default", True)

        try:
            job.validate_data(data)
        except FormsValidationError as e:
            # message_dict can only be accessed if ValidationError got a dict
            # in the constructor (saved as error_dict). Otherwise we get a list
            # of errors under messages
            return Response({"errors": e.message_dict if hasattr(e, "error_dict") else e.messages}, status=400)

        if not get_worker_count():
            raise CeleryWorkerNotRunningException()

        job_content_type = ContentType.objects.get(app_label="extras", model="job")

        schedule = input_serializer.data.get("schedule")
        if schedule:
            schedule = self._create_schedule(schedule, data, commit, job, job_class, request)
        else:
            job_result = JobResult.enqueue_job(
                run_job,
                job.class_path,
                job_content_type,
                request.user,
                data=data,
                request=copy_safe_request(request),
                commit=commit,
            )
            job.result = job_result

        serializer = serializers.JobDetailSerializer(job, context={"request": request})

        return Response(serializer.data)


#
# Job Results
#


class JobLogEntryViewSet(ReadOnlyModelViewSet):
    """
    Retrieve a list of job log entries.
    """

    queryset = JobLogEntry.objects.prefetch_related("job_result")
    serializer_class = serializers.JobLogEntrySerializer
    filterset_class = filters.JobLogEntryFilterSet


class JobResultViewSet(ModelViewSet):
    """
    Retrieve a list of job results
    """

    queryset = JobResult.objects.prefetch_related("user")
    serializer_class = serializers.JobResultSerializer
    filterset_class = filters.JobResultFilterSet

    @action(detail=True)
    def logs(self, request, pk=None):
        job_result = self.get_object()
        logs = job_result.logs.all()
        serializer = nested_serializers.NestedJobLogEntrySerializer(logs, context={"request": request}, many=True)
        return Response(serializer.data)


#
# Scheduled Jobs
#


class ScheduledJobViewSet(ReadOnlyModelViewSet):
    """
    Retrieve a list of scheduled jobs
    """

    queryset = ScheduledJob.objects.prefetch_related("user")
    serializer_class = serializers.ScheduledJobSerializer
    filterset_class = filters.ScheduledJobFilterSet

    @swagger_auto_schema(
        method="post",
        responses={"200": serializers.ScheduledJobSerializer},
        request_body=no_body,
        manual_parameters=[
            openapi.Parameter(
                "force",
                openapi.IN_QUERY,
                description="force execution even if start time has passed",
                type=openapi.TYPE_BOOLEAN,
            )
        ],
    )
    @action(detail=True, methods=["post"])
    def approve(self, request, pk):
        if not request.user.has_perm("extras.run_job"):
            raise PermissionDenied()

        scheduled_job = get_object_or_404(ScheduledJob, pk=pk)

        # Mark the scheduled_job as approved, allowing the schedular to schedule the job execution task
        if request.user == scheduled_job.user:
            # The requestor *cannot* approve their own job
            return Response("You cannot approve your own job request!", status=403)

        if (
            scheduled_job.one_off
            and scheduled_job.start_time < timezone.now()
            and not request.query_params.get("force")
        ):
            return Response(
                "The job's start time is in the past. If you want to force a run anyway, add the `force` query parameter.",
                status=400,
            )

        scheduled_job.approved_by_user = request.user
        scheduled_job.approved_at = timezone.now()
        scheduled_job.save()
        serializer = serializers.ScheduledJobSerializer(scheduled_job, context={"request": request})

        return Response(serializer.data)

    @swagger_auto_schema(
        method="post",
        request_body=no_body,
    )
    @action(detail=True, methods=["post"])
    def deny(self, request, pk):
        if not request.user.has_perm("extras.run_job"):
            raise PermissionDenied()

        scheduled_job = get_object_or_404(ScheduledJob, pk=pk)

        scheduled_job.delete()

        return Response(None)

    @swagger_auto_schema(
        method="post",
        responses={"200": serializers.JobResultSerializer},
        request_body=no_body,
    )
    @action(detail=True, url_path="dry-run", methods=["post"])
    def dry_run(self, request, pk):
        if not request.user.has_perm("extras.run_job"):
            raise PermissionDenied()

        scheduled_job = get_object_or_404(ScheduledJob, pk=pk)
        job_class = get_job(scheduled_job.job_class)
        if job_class is None:
            raise Http404
        job = job_class()
        grouping, module, class_name = job_class.class_path.split("/", 2)

        # Immediately enqueue the job with commit=False
        job_content_type = ContentType.objects.get(app_label="extras", model="job")
        job_result = JobResult.enqueue_job(
            run_job,
            job.class_path,
            job_content_type,
            scheduled_job.user,
            data=scheduled_job.kwargs["data"],
            request=copy_safe_request(request),
            commit=False,  # force a dry-run
        )
        serializer = serializers.JobResultSerializer(job_result, context={"request": request})

        return Response(serializer.data)


#
# Change logging
#


class ObjectChangeViewSet(ReadOnlyModelViewSet):
    """
    Retrieve a list of recent changes.
    """

    metadata_class = ContentTypeMetadata
    queryset = ObjectChange.objects.prefetch_related("user")
    serializer_class = serializers.ObjectChangeSerializer
    filterset_class = filters.ObjectChangeFilterSet


#
#  Relationships
#


class RelationshipViewSet(ModelViewSet):
    metadata_class = ContentTypeMetadata
    queryset = Relationship.objects.all()
    serializer_class = serializers.RelationshipSerializer
    filterset_class = filters.RelationshipFilterSet


class RelationshipAssociationViewSet(ModelViewSet):
    metadata_class = ContentTypeMetadata
    queryset = RelationshipAssociation.objects.all()
    serializer_class = serializers.RelationshipAssociationSerializer
    filterset_class = filters.RelationshipAssociationFilterSet


#
# Secrets
#


class SecretsViewSet(ModelViewSet):
    """
    Manage Secrets through DELETE, GET, POST, PUT, and PATCH requests.
    """

    queryset = Secret.objects.all()
    serializer_class = serializers.SecretSerializer
    filterset_class = filters.SecretFilterSet


class SecretsGroupViewSet(ModelViewSet):
    """
    Manage Secrets Groups through DELETE, GET, POST, PUT, and PATCH requests.
    """

    queryset = SecretsGroup.objects.all()
    serializer_class = serializers.SecretsGroupSerializer
    filterset_class = filters.SecretsGroupFilterSet


class SecretsGroupAssociationViewSet(ModelViewSet):
    """
    Manage Secrets Group Associations through DELETE, GET, POST, PUT, and PATCH requests.
    """

    queryset = SecretsGroupAssociation.objects.all()
    serializer_class = serializers.SecretsGroupAssociationSerializer
    filterset_class = filters.SecretsGroupAssociationFilterSet


#
# Statuses
#


class StatusViewSet(CustomFieldModelViewSet):
    """
    View and manage custom status choices for objects with a `status` field.
    """

    queryset = Status.objects.all()
    serializer_class = serializers.StatusSerializer
    filterset_class = filters.StatusFilterSet


class StatusViewSetMixin(ModelViewSet):
    """
    Mixin to set `metadata_class` to implement `status` field in model viewset metadata.
    """

    metadata_class = StatusFieldMetadata


#
# Tags
#


class TagViewSet(CustomFieldModelViewSet):
    queryset = Tag.objects.annotate(tagged_items=count_related(TaggedItem, "tag"))
    serializer_class = serializers.TagSerializer
    filterset_class = filters.TagFilterSet


#
# Webhooks
#


class WebhooksViewSet(ModelViewSet):
    """
    Manage Webhooks through DELETE, GET, POST, PUT, and PATCH requests.
    """

    queryset = Webhook.objects.all()
    serializer_class = serializers.WebhookSerializer
    filterset_class = filters.WebhookFilterSet

<filename>nautobot/extras/api/views.py
from datetime import datetime
from django.contrib.contenttypes.models import ContentType
from django.forms import ValidationError as FormsValidationError
from django.http import Http404
from django.shortcuts import get_object_or_404
from django.utils import timezone
from drf_yasg import openapi
from drf_yasg.utils import no_body, swagger_auto_schema
from graphene_django.views import GraphQLView
from graphql import GraphQLError
from rest_framework import status
from rest_framework.decorators import action
from rest_framework.exceptions import PermissionDenied
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.routers import APIRootView
from rest_framework import viewsets

from nautobot.core.api.metadata import ContentTypeMetadata, StatusFieldMetadata
from nautobot.core.api.views import ModelViewSet, ReadOnlyModelViewSet
from nautobot.core.graphql import execute_saved_query
from nautobot.extras import filters
from nautobot.extras.choices import JobExecutionType, JobResultStatusChoices
from nautobot.extras.datasources import enqueue_pull_git_repository_and_refresh_data
from nautobot.extras.models import (
    ComputedField,
    ConfigContext,
    ConfigContextSchema,
    CustomLink,
    ExportTemplate,
    GitRepository,
    GraphQLQuery,
    ImageAttachment,
    JobLogEntry,
    JobResult,
    ObjectChange,
    Relationship,
    RelationshipAssociation,
    ScheduledJob,
    Secret,
    SecretsGroup,
    SecretsGroupAssociation,
    Status,
    Tag,
    TaggedItem,
    Webhook,
)
from nautobot.extras.models import CustomField, CustomFieldChoice
from nautobot.extras.jobs import get_job, get_jobs, run_job
from nautobot.extras.utils import get_worker_count
from nautobot.utilities.exceptions import CeleryWorkerNotRunningException
from nautobot.utilities.utils import copy_safe_request, count_related
from . import nested_serializers, serializers


class ExtrasRootView(APIRootView):
    """
    Extras API root view
    """

    def get_view_name(self):
        return "Extras"


#
#  Computed Fields
#


class ComputedFieldViewSet(ModelViewSet):
    """
    Manage Computed Fields through DELETE, GET, POST, PUT, and PATCH requests.
    """

    queryset = ComputedField.objects.all()
    serializer_class = serializers.ComputedFieldSerializer
    filterset_class = filters.ComputedFieldFilterSet


#
# Config contexts
#


class ConfigContextQuerySetMixin:
    """
    Used by views that work with config context models (device and virtual machine).
    Provides a get_queryset() method which deals with adding the config context
    data annotation or not.
    """

    def get_queryset(self):
        """
        Build the proper queryset based on the request context

        If the `brief` query param equates to True or the `exclude` query param
        includes `config_context` as a value, return the base queryset.

        Else, return the queryset annotated with config context data
        """
        queryset = super().get_queryset()
        request = self.get_serializer_context()["request"]
        if self.brief or "config_context" in request.query_params.get("exclude", []):
            return queryset
        return queryset.annotate_config_context_data()


class ConfigContextViewSet(ModelViewSet):
    queryset = ConfigContext.objects.prefetch_related(
        "regions",
        "sites",
        "roles",
        "device_types",
        "platforms",
        "tenant_groups",
        "tenants",
    )
    serializer_class = serializers.ConfigContextSerializer
    filterset_class = filters.ConfigContextFilterSet


#
# Config context schemas
#


class ConfigContextSchemaViewSet(ModelViewSet):
    queryset = ConfigContextSchema.objects.all()
    serializer_class = serializers.ConfigContextSchemaSerializer
    filterset_class = filters.ConfigContextSchemaFilterSet


#
# ContentTypes
#


class ContentTypeViewSet(viewsets.ReadOnlyModelViewSet):
    """
    Read-only list of ContentTypes. Limit results to ContentTypes pertinent to Nautobot objects.
    """

    queryset = ContentType.objects.order_by("app_label", "model")
    serializer_class = serializers.ContentTypeSerializer
    filterset_class = filters.ContentTypeFilterSet


#
# Custom fields
#


class CustomFieldViewSet(ModelViewSet):
    metadata_class = ContentTypeMetadata
    queryset = CustomField.objects.all()
    serializer_class = serializers.CustomFieldSerializer
    filterset_class = filters.CustomFieldFilterSet


class CustomFieldChoiceViewSet(ModelViewSet):
    queryset = CustomFieldChoice.objects.all()
    serializer_class = serializers.CustomFieldChoiceSerializer
    filterset_class = filters.CustomFieldChoiceFilterSet


class CustomFieldModelViewSet(ModelViewSet):
    """
    Include the applicable set of CustomFields in the ModelViewSet context.
    """

    def get_serializer_context(self):

        # Gather all custom fields for the model
        content_type = ContentType.objects.get_for_model(self.queryset.model)
        custom_fields = content_type.custom_fields.all()

        context = super().get_serializer_context()
        context.update(
            {
                "custom_fields": custom_fields,
            }
        )
        return context


#
# Custom Links
#


class CustomLinkViewSet(ModelViewSet):
    """
    Manage Custom Links through DELETE, GET, POST, PUT, and PATCH requests.
    """

    queryset = CustomLink.objects.all()
    serializer_class = serializers.CustomLinkSerializer
    filterset_class = filters.CustomLinkFilterSet


#
# Export templates
#


class ExportTemplateViewSet(ModelViewSet):
    metadata_class = ContentTypeMetadata
    queryset = ExportTemplate.objects.all()
    serializer_class = serializers.ExportTemplateSerializer
    filterset_class = filters.ExportTemplateFilterSet


#
# Git repositories
#


class GitRepositoryViewSet(CustomFieldModelViewSet):
    """
    Manage the use of Git repositories as external data sources.
    """

    queryset = GitRepository.objects.all()
    serializer_class = serializers.GitRepositorySerializer
    filterset_class = filters.GitRepositoryFilterSet

    @swagger_auto_schema(method="post", request_body=serializers.GitRepositorySerializer)
    @action(detail=True, methods=["post"])
    def sync(self, request, pk):
        """
        Enqueue pull git repository and refresh data.
        """
        if not request.user.has_perm("extras.change_gitrepository"):
            raise PermissionDenied("This user does not have permission to make changes to Git repositories.")

        if not get_worker_count():
            raise CeleryWorkerNotRunningException()

        repository = get_object_or_404(GitRepository, id=pk)
        enqueue_pull_git_repository_and_refresh_data(repository, request)
        return Response({"message": f"Repository {repository} sync job added to queue."})


#
# GraphQL Queries
#


class GraphQLQueryViewSet(ModelViewSet):
    queryset = GraphQLQuery.objects.all()
    serializer_class = serializers.GraphQLQuerySerializer
    filterset_class = filters.GraphQLQueryFilterSet

    @swagger_auto_schema(
        method="post",
        request_body=serializers.GraphQLQueryInputSerializer,
        responses={"200": serializers.GraphQLQueryOutputSerializer},
    )
    @action(detail=True, methods=["post"])
    def run(self, request, pk):
        try:
            query = get_object_or_404(self.queryset, pk=pk)
            result = execute_saved_query(query.slug, variables=request.data.get("variables"), request=request).to_dict()
            return Response(result)
        except GraphQLError as error:
            return Response(
                {"errors": [GraphQLView.format_error(error)]},
                status=status.HTTP_400_BAD_REQUEST,
            )


#
# Image attachments
#


class ImageAttachmentViewSet(ModelViewSet):
    metadata_class = ContentTypeMetadata
    queryset = ImageAttachment.objects.all()
    serializer_class = serializers.ImageAttachmentSerializer
    filterset_class = filters.ImageAttachmentFilterSet


#
# Jobs
#


class JobViewSet(viewsets.ViewSet):
    permission_classes = [IsAuthenticated]
    lookup_field = "class_path"
    lookup_value_regex = "[^/]+/[^/]+/[^/]+"  # e.g. "git.repo_name/module_name/JobName"

    def _get_job_class(self, class_path):
        job_class = get_job(class_path)
        if job_class is None:
            raise Http404

        return job_class

    def _create_schedule(self, serializer, data, commit, job, job_class, request):
        """
        This is an internal function to create a scheduled job from API data.
        It has to handle boths once-offs (i.e. of type TYPE_FUTURE) and interval
        jobs.
        """
        job_kwargs = {
            "data": data,
            "request": copy_safe_request(request),
            "user": request.user.pk,
            "commit": commit,
            "name": job.class_path,
        }
        type_ = serializer["interval"]
        if type_ == JobExecutionType.TYPE_IMMEDIATELY:
            time = datetime.now()
            name = serializer.get("name") or f"{job.name} - {time}"
        else:
            time = serializer["start_time"]
            name = serializer["name"]
        scheduled_job = ScheduledJob(
            name=name,
            task="nautobot.extras.jobs.scheduled_job_handler",
            job_class=job.class_path,
            start_time=time,
            description=f"Nautobot job {name} scheduled by {request.user} on {time}",
            kwargs=job_kwargs,
            interval=type_,
            one_off=(type_ == JobExecutionType.TYPE_FUTURE),
            user=request.user,
            approval_required=job_class.approval_required,
        )
        scheduled_job.save()
        return scheduled_job

    def list(self, request):
        if not request.user.has_perm("extras.view_job"):
            raise PermissionDenied("This user does not have permission to view jobs.")
        job_content_type = ContentType.objects.get(app_label="extras", model="job")
        results = {
            r.name: r
            for r in JobResult.objects.filter(
                obj_type=job_content_type,
                status__in=JobResultStatusChoices.TERMINAL_STATE_CHOICES,
            )
            .defer("data")
            .order_by("created")
        }

        jobs = get_jobs()
        jobs_list = []
        for grouping, modules in jobs.items():
            for module_name, entry in modules.items():
                for job_class in entry["jobs"].values():
                    job = job_class()
                    job.result = results.get(job.class_path, None)
                    jobs_list.append(job)

        serializer = serializers.JobSerializer(jobs_list, many=True, context={"request": request})

        return Response(serializer.data)

    def retrieve(self, request, class_path):
        if not request.user.has_perm("extras.view_job"):
            raise PermissionDenied("This user does not have permission to view jobs.")
        job_class = self._get_job_class(class_path)
        job_content_type = ContentType.objects.get(app_label="extras", model="job")
        job = job_class()
        job.result = JobResult.objects.filter(
            obj_type=job_content_type,
            name=job.class_path,
            status__in=JobResultStatusChoices.TERMINAL_STATE_CHOICES,
        ).first()

        serializer = serializers.JobDetailSerializer(job, context={"request": request})

        return Response(serializer.data)

    @swagger_auto_schema(method="post", request_body=serializers.JobInputSerializer)
    @action(detail=True, methods=["post"])
    def run(self, request, class_path):
        if not request.user.has_perm("extras.run_job"):
            raise PermissionDenied("This user does not have permission to run jobs.")

        job_class = self._get_job_class(class_path)
        job = job_class()

        input_serializer = serializers.JobInputSerializer(data=request.data)
        input_serializer.is_valid(raise_exception=True)

        data = input_serializer.data["data"] or {}
        commit = input_serializer.data["commit"]
        if commit is None:
            commit = getattr(job_class.Meta, "commit_default", True)

        try:
            job.validate_data(data)
        except FormsValidationError as e:
            # message_dict can only be accessed if ValidationError got a dict
            # in the constructor (saved as error_dict). Otherwise we get a list
            # of errors under messages
            return Response({"errors": e.message_dict if hasattr(e, "error_dict") else e.messages}, status=400)

        if not get_worker_count():
            raise CeleryWorkerNotRunningException()

        job_content_type = ContentType.objects.get(app_label="extras", model="job")

        schedule = input_serializer.data.get("schedule")
        if schedule:
            schedule = self._create_schedule(schedule, data, commit, job, job_class, request)
        else:
            job_result = JobResult.enqueue_job(
                run_job,
                job.class_path,
                job_content_type,
                request.user,
                data=data,
                request=copy_safe_request(request),
                commit=commit,
            )
            job.result = job_result

        serializer = serializers.JobDetailSerializer(job, context={"request": request})

        return Response(serializer.data)


#
# Job Results
#


class JobLogEntryViewSet(ReadOnlyModelViewSet):
    """
    Retrieve a list of job log entries.
    """

    queryset = JobLogEntry.objects.prefetch_related("job_result")
    serializer_class = serializers.JobLogEntrySerializer
    filterset_class = filters.JobLogEntryFilterSet


class JobResultViewSet(ModelViewSet):
    """
    Retrieve a list of job results
    """

    queryset = JobResult.objects.prefetch_related("user")
    serializer_class = serializers.JobResultSerializer
    filterset_class = filters.JobResultFilterSet

    @action(detail=True)
    def logs(self, request, pk=None):
        job_result = self.get_object()
        logs = job_result.logs.all()
        serializer = nested_serializers.NestedJobLogEntrySerializer(logs, context={"request": request}, many=True)
        return Response(serializer.data)


#
# Scheduled Jobs
#


class ScheduledJobViewSet(ReadOnlyModelViewSet):
    """
    Retrieve a list of scheduled jobs
    """

    queryset = ScheduledJob.objects.prefetch_related("user")
    serializer_class = serializers.ScheduledJobSerializer
    filterset_class = filters.ScheduledJobFilterSet

    @swagger_auto_schema(
        method="post",
        responses={"200": serializers.ScheduledJobSerializer},
        request_body=no_body,
        manual_parameters=[
            openapi.Parameter(
                "force",
                openapi.IN_QUERY,
                description="force execution even if start time has passed",
                type=openapi.TYPE_BOOLEAN,
            )
        ],
    )
    @action(detail=True, methods=["post"])
    def approve(self, request, pk):
        if not request.user.has_perm("extras.run_job"):
            raise PermissionDenied()

        scheduled_job = get_object_or_404(ScheduledJob, pk=pk)

        # Mark the scheduled_job as approved, allowing the schedular to schedule the job execution task
        if request.user == scheduled_job.user:
            # The requestor *cannot* approve their own job
            return Response("You cannot approve your own job request!", status=403)

        if (
            scheduled_job.one_off
            and scheduled_job.start_time < timezone.now()
            and not request.query_params.get("force")
        ):
            return Response(
                "The job's start time is in the past. If you want to force a run anyway, add the `force` query parameter.",
                status=400,
            )

        scheduled_job.approved_by_user = request.user
        scheduled_job.approved_at = timezone.now()
        scheduled_job.save()
        serializer = serializers.ScheduledJobSerializer(scheduled_job, context={"request": request})

        return Response(serializer.data)

    @swagger_auto_schema(
        method="post",
        request_body=no_body,
    )
    @action(detail=True, methods=["post"])
    def deny(self, request, pk):
        if not request.user.has_perm("extras.run_job"):
            raise PermissionDenied()

        scheduled_job = get_object_or_404(ScheduledJob, pk=pk)

        scheduled_job.delete()

        return Response(None)

    @swagger_auto_schema(
        method="post",
        responses={"200": serializers.JobResultSerializer},
        request_body=no_body,
    )
    @action(detail=True, url_path="dry-run", methods=["post"])
    def dry_run(self, request, pk):
        if not request.user.has_perm("extras.run_job"):
            raise PermissionDenied()

        scheduled_job = get_object_or_404(ScheduledJob, pk=pk)
        job_class = get_job(scheduled_job.job_class)
        if job_class is None:
            raise Http404
        job = job_class()
        grouping, module, class_name = job_class.class_path.split("/", 2)

        # Immediately enqueue the job with commit=False
        job_content_type = ContentType.objects.get(app_label="extras", model="job")
        job_result = JobResult.enqueue_job(
            run_job,
            job.class_path,
            job_content_type,
            scheduled_job.user,
            data=scheduled_job.kwargs["data"],
            request=copy_safe_request(request),
            commit=False,  # force a dry-run
        )
        serializer = serializers.JobResultSerializer(job_result, context={"request": request})

        return Response(serializer.data)


#
# Change logging
#


class ObjectChangeViewSet(ReadOnlyModelViewSet):
    """
    Retrieve a list of recent changes.
    """

    metadata_class = ContentTypeMetadata
    queryset = ObjectChange.objects.prefetch_related("user")
    serializer_class = serializers.ObjectChangeSerializer
    filterset_class = filters.ObjectChangeFilterSet


#
#  Relationships
#


class RelationshipViewSet(ModelViewSet):
    metadata_class = ContentTypeMetadata
    queryset = Relationship.objects.all()
    serializer_class = serializers.RelationshipSerializer
    filterset_class = filters.RelationshipFilterSet


class RelationshipAssociationViewSet(ModelViewSet):
    metadata_class = ContentTypeMetadata
    queryset = RelationshipAssociation.objects.all()
    serializer_class = serializers.RelationshipAssociationSerializer
    filterset_class = filters.RelationshipAssociationFilterSet


#
# Secrets
#


class SecretsViewSet(ModelViewSet):
    """
    Manage Secrets through DELETE, GET, POST, PUT, and PATCH requests.
    """

    queryset = Secret.objects.all()
    serializer_class = serializers.SecretSerializer
    filterset_class = filters.SecretFilterSet


class SecretsGroupViewSet(ModelViewSet):
    """
    Manage Secrets Groups through DELETE, GET, POST, PUT, and PATCH requests.
    """

    queryset = SecretsGroup.objects.all()
    serializer_class = serializers.SecretsGroupSerializer
    filterset_class = filters.SecretsGroupFilterSet


class SecretsGroupAssociationViewSet(ModelViewSet):
    """
    Manage Secrets Group Associations through DELETE, GET, POST, PUT, and PATCH requests.
    """

    queryset = SecretsGroupAssociation.objects.all()
    serializer_class = serializers.SecretsGroupAssociationSerializer
    filterset_class = filters.SecretsGroupAssociationFilterSet


#
# Statuses
#


class StatusViewSet(CustomFieldModelViewSet):
    """
    View and manage custom status choices for objects with a `status` field.
    """

    queryset = Status.objects.all()
    serializer_class = serializers.StatusSerializer
    filterset_class = filters.StatusFilterSet


class StatusViewSetMixin(ModelViewSet):
    """
    Mixin to set `metadata_class` to implement `status` field in model viewset metadata.
    """

    metadata_class = StatusFieldMetadata


#
# Tags
#


class TagViewSet(CustomFieldModelViewSet):
    queryset = Tag.objects.annotate(tagged_items=count_related(TaggedItem, "tag"))
    serializer_class = serializers.TagSerializer
    filterset_class = filters.TagFilterSet


#
# Webhooks
#


class WebhooksViewSet(ModelViewSet):
    """
    Manage Webhooks through DELETE, GET, POST, PUT, and PATCH requests.
    """

    queryset = Webhook.objects.all()
    serializer_class = serializers.WebhookSerializer
    filterset_class = filters.WebhookFilterSet

Extras API root view # #  Computed Fields # Manage Computed Fields through DELETE, GET, POST, PUT, and PATCH requests. # # Config contexts # Used by views that work with config context models (device and virtual machine).     Provides a get_queryset() method which deals with adding the config context     data annotation or not. Build the proper queryset based on the request context          If the `brief` query param equates to True or the `exclude` query param         includes `config_context` as a value, return the base queryset.          Else, return the queryset annotated with config context data # # Config context schemas # # # ContentTypes # Read-only list of ContentTypes. Limit results to ContentTypes pertinent to Nautobot objects. # # Custom fields # Include the applicable set of CustomFields in the ModelViewSet context. # Gather all custom fields for the model # # Custom Links # Manage Custom Links through DELETE, GET, POST, PUT, and PATCH requests. # # Export templates # # # Git repositories # Manage the use of Git repositories as external data sources. Enqueue pull git repository and refresh data. # # GraphQL Queries # # # Image attachments # # # Jobs # # e.g. "git.repo_name/module_name/JobName" This is an internal function to create a scheduled job from API data.         It has to handle boths once-offs (i.e. of type TYPE_FUTURE) and interval         jobs. # message_dict can only be accessed if ValidationError got a dict # in the constructor (saved as error_dict). Otherwise we get a list # of errors under messages # # Job Results # Retrieve a list of job log entries. Retrieve a list of job results # # Scheduled Jobs # Retrieve a list of scheduled jobs # Mark the scheduled_job as approved, allowing the schedular to schedule the job execution task # The requestor *cannot* approve their own job # Immediately enqueue the job with commit=False # force a dry-run # # Change logging # Retrieve a list of recent changes. # #  Relationships # # # Secrets # Manage Secrets through DELETE, GET, POST, PUT, and PATCH requests. Manage Secrets Groups through DELETE, GET, POST, PUT, and PATCH requests. Manage Secrets Group Associations through DELETE, GET, POST, PUT, and PATCH requests. # # Statuses # View and manage custom status choices for objects with a `status` field. Mixin to set `metadata_class` to implement `status` field in model viewset metadata. # # Tags # # # Webhooks # Manage Webhooks through DELETE, GET, POST, PUT, and PATCH requests.