mirror/netbox-community-netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2024-05-10 07:54:54 +00:00
Code Activity

Closes #14438: Database representation of scripts

Browse Source 
- Introduces the Script model to represent individual Python classes within a ScriptModule file
- Automatically migrates jobs & event rules

---------

Co-authored-by: Jeremy Stretch <jstretch@netboxlabs.com>
This commit is contained in:
Arthur Hanson
2024-02-23 05:27:37 -08:00
committed by GitHub
parent 7e7e5d5eb0
commit ca2ee436a0
25 changed files with 569 additions and 336 deletions
Download Patch File Download Diff File Expand all files Collapse all files

102
netbox/extras/api/serializers.py
View File

@@ -44,9 +44,6 @@ __all__ = (
'ImageAttachmentSerializer',
'JournalEntrySerializer',
'ObjectChangeSerializer',
'ReportDetailSerializer',
'ReportSerializer',
'ReportInputSerializer',
'SavedFilterSerializer',
'ScriptDetailSerializer',
'ScriptInputSerializer',
@@ -85,9 +82,9 @@ class EventRuleSerializer(NetBoxModelSerializer):
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 ConfigTemplateSerializer(TaggableModelSerializer, ValidatedModelSerializer
]
#
# 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):

64
netbox/extras/api/views.py
View File

@@ -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.generics import RetrieveUpdateDestroyAPIView
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 @@ class ConfigTemplateViewSet(SyncedDataMixin, ConfigTemplateRenderMixin, NetBoxMo
# 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 @@ class ScriptViewSet(ViewSet):
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')
)