14438 database representation of scripts

netbox/extras/api/serializers.py

@@ -44,9 +44,6 @@
     'ImageAttachmentSerializer',
     'JournalEntrySerializer',
     'ObjectChangeSerializer',
-    'ReportDetailSerializer',
-    'ReportSerializer',
-    'ReportInputSerializer',
     'SavedFilterSerializer',
     'ScriptDetailSerializer',
     'ScriptInputSerializer',
@@ -85,9 +82,9 @@ def get_action_object(self, instance):
         context = {'request': self.context['request']}
         # We need to manually instantiate the serializer for scripts
         if instance.action_type == EventRuleActionChoices.SCRIPT:
-            script_name = instance.action_parameters['script_name']
-            script = instance.action_object.scripts[script_name]()
-            return NestedScriptSerializer(script, context=context).data
+            script = instance.action_object
+            instance = script.python_class() if script.python_class else None
+            return NestedScriptSerializer(instance, context=context).data
         else:
             serializer = get_serializer_for_model(
                 model=instance.action_object_type.model_class(),
@@ -512,79 +509,54 @@ class Meta:
         ]
 
 
-#
-# Reports
-#
-
-class ReportSerializer(serializers.Serializer):
-    url = serializers.HyperlinkedIdentityField(
-        view_name='extras-api:report-detail',
-        lookup_field='full_name',
-        lookup_url_kwarg='pk'
-    )
-    id = serializers.CharField(read_only=True, source="full_name")
-    module = serializers.CharField(max_length=255)
-    name = serializers.CharField(max_length=255)
-    description = serializers.CharField(max_length=255, required=False)
-    test_methods = serializers.ListField(child=serializers.CharField(max_length=255), read_only=True)
-    result = NestedJobSerializer()
-    display = serializers.SerializerMethodField(read_only=True)
-
-    @extend_schema_field(serializers.CharField())
-    def get_display(self, obj):
-        return f'{obj.name} ({obj.module})'
-
-
-class ReportDetailSerializer(ReportSerializer):
-    result = JobSerializer()
-
-
-class ReportInputSerializer(serializers.Serializer):
-    schedule_at = serializers.DateTimeField(required=False, allow_null=True)
-    interval = serializers.IntegerField(required=False, allow_null=True)
-
-    def validate_schedule_at(self, value):
-        if value and not self.context['report'].scheduling_enabled:
-            raise serializers.ValidationError(_("Scheduling is not enabled for this report."))
-        return value
-
-    def validate_interval(self, value):
-        if value and not self.context['report'].scheduling_enabled:
-            raise serializers.ValidationError(_("Scheduling is not enabled for this report."))
-        return value
-
-
 #
 # Scripts
 #
 
-class ScriptSerializer(serializers.Serializer):
-    url = serializers.HyperlinkedIdentityField(
-        view_name='extras-api:script-detail',
-        lookup_field='full_name',
-        lookup_url_kwarg='pk'
-    )
-    id = serializers.CharField(read_only=True, source="full_name")
-    module = serializers.CharField(max_length=255)
-    name = serializers.CharField(read_only=True)
-    description = serializers.CharField(read_only=True)
+class ScriptSerializer(ValidatedModelSerializer):
+    url = serializers.HyperlinkedIdentityField(view_name='extras-api:script-detail')
+    description = serializers.SerializerMethodField(read_only=True)
     vars = serializers.SerializerMethodField(read_only=True)
-    result = NestedJobSerializer()
-    display = serializers.SerializerMethodField(read_only=True)
+    result = NestedJobSerializer(read_only=True)
+
+    class Meta:
+        model = Script
+        fields = [
+            'id', 'url', 'module', 'name', 'description', 'vars', 'result', 'display', 'is_executable',
+        ]
 
     @extend_schema_field(serializers.JSONField(allow_null=True))
-    def get_vars(self, instance):
-        return {
-            k: v.__class__.__name__ for k, v in instance._get_vars().items()
-        }
+    def get_vars(self, obj):
+        if obj.python_class:
+            return {
+                k: v.__class__.__name__ for k, v in obj.python_class()._get_vars().items()
+            }
+        else:
+            return {}
 
     @extend_schema_field(serializers.CharField())
     def get_display(self, obj):
         return f'{obj.name} ({obj.module})'
 
+    @extend_schema_field(serializers.CharField())
+    def get_description(self, obj):
+        if obj.python_class:
+            return obj.python_class().description
+        else:
+            return None
+
 
 class ScriptDetailSerializer(ScriptSerializer):
-    result = JobSerializer()
+    result = serializers.SerializerMethodField(read_only=True)
+
+    @extend_schema_field(JobSerializer())
+    def get_result(self, obj):
+        job = obj.jobs.all().order_by('-created').first()
+        context = {
+            'request': self.context['request']
+        }
+        data = JobSerializer(job, context=context).data
+        return data
 
 
 class ScriptInputSerializer(serializers.Serializer):

netbox/extras/api/views.py

@@ -1,5 +1,4 @@
 from django.contrib.contenttypes.models import ContentType
-from django.http import Http404
 from django.shortcuts import get_object_or_404
 from django_rq.queues import get_connection
 from rest_framework import status
@@ -9,14 +8,13 @@
 from rest_framework.renderers import JSONRenderer
 from rest_framework.response import Response
 from rest_framework.routers import APIRootView
-from rest_framework.viewsets import ReadOnlyModelViewSet, ViewSet
+from rest_framework.viewsets import ModelViewSet, ReadOnlyModelViewSet
 from rq import Worker
 
-from core.choices import JobStatusChoices
 from core.models import Job
 from extras import filtersets
 from extras.models import *
-from extras.scripts import get_module_and_script, run_script
+from extras.scripts import run_script
 from netbox.api.authentication import IsAuthenticatedOrLoginNotRequired
 from netbox.api.features import SyncedDataMixin
 from netbox.api.metadata import ContentTypeMetadata
@@ -209,66 +207,30 @@ def render(self, request, pk):
 # Scripts
 #
 
-class ScriptViewSet(ViewSet):
+class ScriptViewSet(ModelViewSet):
     permission_classes = [IsAuthenticatedOrLoginNotRequired]
+    queryset = Script.objects.prefetch_related('jobs')
+    serializer_class = serializers.ScriptSerializer
+    filterset_class = filtersets.ScriptFilterSet
+
     _ignore_model_permissions = True
-    schema = None
     lookup_value_regex = '[^/]+'  # Allow dots
 
-    def _get_script(self, pk):
-        try:
-            module_name, script_name = pk.split('.', maxsplit=1)
-        except ValueError:
-            raise Http404
-
-        module, script = get_module_and_script(module_name, script_name)
-        if script is None:
-            raise Http404
-
-        return module, script
-
-    def list(self, request):
-        results = {
-            job.name: job
-            for job in Job.objects.filter(
-                object_type=ContentType.objects.get(app_label='extras', model='scriptmodule'),
-                status__in=JobStatusChoices.TERMINAL_STATE_CHOICES
-            ).order_by('name', '-created').distinct('name').defer('data')
-        }
-
-        script_list = []
-        for script_module in ScriptModule.objects.restrict(request.user):
-            script_list.extend(script_module.scripts.values())
-
-        # Attach Job objects to each script (if any)
-        for script in script_list:
-            script.result = results.get(script.class_name, None)
-
-        serializer = serializers.ScriptSerializer(script_list, many=True, context={'request': request})
-
-        return Response({'count': len(script_list), 'results': serializer.data})
-
     def retrieve(self, request, pk):
-        module, script = self._get_script(pk)
-        object_type = ContentType.objects.get(app_label='extras', model='scriptmodule')
-        script.result = Job.objects.filter(
-            object_type=object_type,
-            name=script.class_name,
-            status__in=JobStatusChoices.TERMINAL_STATE_CHOICES
-        ).first()
+        script = get_object_or_404(self.queryset, pk=pk)
         serializer = serializers.ScriptDetailSerializer(script, context={'request': request})
 
         return Response(serializer.data)
 
     def post(self, request, pk):
         """
-        Run a Script identified as "<module>.<script>" and return the pending Job as the result
+        Run a Script identified by the id and return the pending Job as the result
         """
 
         if not request.user.has_perm('extras.run_script'):
             raise PermissionDenied("This user does not have permission to run scripts.")
 
-        module, script = self._get_script(pk)
+        script = get_object_or_404(self.queryset, pk=pk)
         input_serializer = serializers.ScriptInputSerializer(
             data=request.data,
             context={'script': script}
@@ -281,13 +243,13 @@ def post(self, request, pk):
         if input_serializer.is_valid():
             script.result = Job.enqueue(
                 run_script,
-                instance=module,
-                name=script.class_name,
+                instance=script.module,
+                name=script.python_class.class_name,
                 user=request.user,
                 data=input_serializer.data['data'],
                 request=copy_safe_request(request),
                 commit=input_serializer.data['commit'],
-                job_timeout=script.job_timeout,
+                job_timeout=script.python_class.job_timeout,
                 schedule_at=input_serializer.validated_data.get('schedule_at'),
                 interval=input_serializer.validated_data.get('interval')
             )

netbox/extras/events.py

@@ -116,15 +116,13 @@ def process_event_rules(event_rules, model_name, event, data, username=None, sna
         # Scripts
         elif event_rule.action_type == EventRuleActionChoices.SCRIPT:
             # Resolve the script from action parameters
-            script_module = event_rule.action_object
-            script_name = event_rule.action_parameters['script_name']
-            script = script_module.scripts[script_name]()
+            script = event_rule.action_object.python_class()
 
             # Enqueue a Job to record the script's execution
             Job.enqueue(
                 "extras.scripts.run_script",
-                instance=script_module,
-                name=script.class_name,
+                instance=script.module,
+                name=script.name,
                 user=user,
                 data=data
             )

netbox/extras/filtersets.py

@@ -29,11 +29,32 @@
     'LocalConfigContextFilterSet',
     'ObjectChangeFilterSet',
     'SavedFilterFilterSet',
+    'ScriptFilterSet',
     'TagFilterSet',
     'WebhookFilterSet',
 )
 
 
+class ScriptFilterSet(BaseFilterSet):
+    q = django_filters.CharFilter(
+        method='search',
+        label=_('Search'),
+    )
+
+    class Meta:
+        model = Script
+        fields = [
+            'id', 'name',
+        ]
+
+    def search(self, queryset, name, value):
+        if not value.strip():
+            return queryset
+        return queryset.filter(
+            Q(name__icontains=value)
+        )
+
+
 class WebhookFilterSet(NetBoxModelFilterSet):
     q = django_filters.CharFilter(
         method='search',

netbox/extras/forms/bulk_import.py

@@ -212,11 +212,8 @@ def clean(self):
                     module, script = get_module_and_script(module_name, script_name)
                 except ObjectDoesNotExist:
                     raise forms.ValidationError(_("Script {name} not found").format(name=action_object))
-                self.instance.action_object = module
-                self.instance.action_object_type = ContentType.objects.get_for_model(module, for_concrete_model=False)
-                self.instance.action_parameters = {
-                    'script_name': script_name,
-                }
+                self.instance.action_object = script
+                self.instance.action_object_type = ContentType.objects.get_for_model(script, for_concrete_model=False)
 
 
 class TagImportForm(CSVModelForm):

netbox/extras/forms/model_forms.py

@@ -297,20 +297,16 @@ class Meta:
         }
 
     def init_script_choice(self):
-        choices = []
-        for module in ScriptModule.objects.all():
-            scripts = []
-            for script_name in module.scripts.keys():
-                name = f"{str(module.pk)}:{script_name}"
-                scripts.append((name, script_name))
-            if scripts:
-                choices.append((str(module), scripts))
-        self.fields['action_choice'].choices = choices
-
-        if self.instance.action_type == EventRuleActionChoices.SCRIPT and self.instance.action_parameters:
-            scriptmodule_id = self.instance.action_object_id
-            script_name = self.instance.action_parameters.get('script_name')
-            self.fields['action_choice'].initial = f'{scriptmodule_id}:{script_name}'
+        initial = None
+        if self.instance.action_type == EventRuleActionChoices.SCRIPT:
+            script_id = get_field_value(self, 'action_object_id')
+            initial = Script.objects.get(pk=script_id) if script_id else None
+        self.fields['action_choice'] = DynamicModelChoiceField(
+            label=_('Script'),
+            queryset=Script.objects.all(),
+            required=True,
+            initial=initial
+        )
 
     def init_webhook_choice(self):
         initial = None
@@ -348,26 +344,13 @@ def clean(self):
         # Script
         elif self.cleaned_data.get('action_type') == EventRuleActionChoices.SCRIPT:
             self.cleaned_data['action_object_type'] = ContentType.objects.get_for_model(
-                ScriptModule,
+                Script,
                 for_concrete_model=False
             )
-            module_id, script_name = action_choice.split(":", maxsplit=1)
-            self.cleaned_data['action_object_id'] = module_id
+            self.cleaned_data['action_object_id'] = action_choice.id
 
         return self.cleaned_data
 
-    def save(self, *args, **kwargs):
-        # Set action_parameters on the instance
-        if self.cleaned_data['action_type'] == EventRuleActionChoices.SCRIPT:
-            module_id, script_name = self.cleaned_data.get('action_choice').split(":", maxsplit=1)
-            self.instance.action_parameters = {
-                'script_name': script_name,
-            }
-        else:
-            self.instance.action_parameters = None
-
-        return super().save(*args, **kwargs)
-
 
 class TagForm(forms.ModelForm):
     slug = SlugField()