from django.contrib.contenttypes.models import ContentType from django.db.models import Count from django.http import Http404 from django_rq.queues import get_connection from rest_framework import status from rest_framework.decorators import action from rest_framework.exceptions import PermissionDenied from rest_framework.response import Response from rest_framework.routers import APIRootView from rest_framework.viewsets import ReadOnlyModelViewSet, ViewSet from rq import Worker from extras import filters from extras.choices import JobResultStatusChoices from extras.models import ConfigContext, ExportTemplate, ImageAttachment, ObjectChange, JobResult, Tag from extras.reports import get_report, get_reports, run_report from extras.scripts import get_script, get_scripts, run_script from netbox.api.views import ModelViewSet from netbox.api.authentication import IsAuthenticatedOrLoginNotRequired from netbox.api.metadata import ContentTypeMetadata from utilities.exceptions import RQWorkerNotRunningException from utilities.utils import copy_safe_request from . import serializers class ExtrasRootView(APIRootView): """ Extras API root view """ def get_view_name(self): return 'Extras' # # Custom fields # 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 # # Export templates # class ExportTemplateViewSet(ModelViewSet): metadata_class = ContentTypeMetadata queryset = ExportTemplate.objects.all() serializer_class = serializers.ExportTemplateSerializer filterset_class = filters.ExportTemplateFilterSet # # Tags # class TagViewSet(ModelViewSet): queryset = Tag.objects.annotate( tagged_items=Count('extras_taggeditem_items') ).order_by(*Tag._meta.ordering) serializer_class = serializers.TagSerializer filterset_class = filters.TagFilterSet # # Image attachments # class ImageAttachmentViewSet(ModelViewSet): metadata_class = ContentTypeMetadata queryset = ImageAttachment.objects.all() serializer_class = serializers.ImageAttachmentSerializer filterset_class = filters.ImageAttachmentFilterSet # # Config contexts # class ConfigContextViewSet(ModelViewSet): queryset = ConfigContext.objects.prefetch_related( 'regions', 'sites', 'roles', 'platforms', 'tenant_groups', 'tenants', ) serializer_class = serializers.ConfigContextSerializer filterset_class = filters.ConfigContextFilterSet # # Reports # class ReportViewSet(ViewSet): permission_classes = [IsAuthenticatedOrLoginNotRequired] _ignore_model_permissions = True exclude_from_schema = True lookup_value_regex = '[^/]+' # Allow dots def _retrieve_report(self, pk): # Read the PK as "." if '.' not in pk: raise Http404 module_name, report_name = pk.split('.', 1) # Raise a 404 on an invalid Report module/name report = get_report(module_name, report_name) if report is None: raise Http404 return report def list(self, request): """ Compile all reports and their related results (if any). Result data is deferred in the list view. """ report_list = [] report_content_type = ContentType.objects.get(app_label='extras', model='report') results = { r.name: r for r in JobResult.objects.filter( obj_type=report_content_type, status__in=JobResultStatusChoices.TERMINAL_STATE_CHOICES ).defer('data') } # Iterate through all available Reports. for module_name, reports in get_reports(): for report in reports: # Attach the relevant JobResult (if any) to each Report. report.result = results.get(report.full_name, None) report_list.append(report) serializer = serializers.ReportSerializer(report_list, many=True, context={ 'request': request, }) return Response(serializer.data) def retrieve(self, request, pk): """ Retrieve a single Report identified as ".". """ # Retrieve the Report and JobResult, if any. report = self._retrieve_report(pk) report_content_type = ContentType.objects.get(app_label='extras', model='report') report.result = JobResult.objects.filter( obj_type=report_content_type, name=report.full_name, status__in=JobResultStatusChoices.TERMINAL_STATE_CHOICES ).first() serializer = serializers.ReportDetailSerializer(report, context={ 'request': request }) return Response(serializer.data) @action(detail=True, methods=['post']) def run(self, request, pk): """ Run a Report identified as ".