Coverage for core/api/views.py: 81%
158 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1from django.http import Http404, HttpResponse
2from django.shortcuts import get_object_or_404
3from django.utils.translation import gettext_lazy as _
4from django_rq.queues import get_redis_connection
5from django_rq.settings import get_queues_list
6from django_rq.utils import get_statistics
7from drf_spectacular.types import OpenApiTypes
8from drf_spectacular.utils import OpenApiParameter, extend_schema
9from rest_framework import viewsets
10from rest_framework.decorators import action
11from rest_framework.exceptions import PermissionDenied
12from rest_framework.response import Response
13from rest_framework.routers import APIRootView
14from rq.job import Job as RQ_Job
15from rq.worker import Worker
17from core import filtersets
18from core.jobs import SyncDataSourceJob
19from core.models import *
20from core.utils import delete_rq_job, enqueue_rq_job, get_rq_jobs, requeue_rq_job, stop_rq_job
21from netbox.api.authentication import IsAuthenticatedOrLoginNotRequired
22from netbox.api.metadata import ContentTypeMetadata
23from netbox.api.pagination import LimitOffsetListPagination
24from netbox.api.viewsets import NetBoxModelViewSet, NetBoxReadOnlyModelViewSet
25from utilities.api import IsSuperuser
27from . import serializers
30class CoreRootView(APIRootView):
31 """
32 Core API root view
33 """
34 def get_view_name(self):
35 return 'Core'
38class DataSourceViewSet(NetBoxModelViewSet):
39 queryset = DataSource.objects.all()
40 serializer_class = serializers.DataSourceSerializer
41 filterset_class = filtersets.DataSourceFilterSet
43 @action(detail=True, methods=['post'])
44 def sync(self, request, pk):
45 """
46 Enqueue a job to synchronize the DataSource.
47 """
48 datasource = get_object_or_404(DataSource, pk=pk)
50 if not request.user.has_perm('core.sync_datasource', obj=datasource):
51 raise PermissionDenied(_("This user does not have permission to synchronize this data source."))
53 # Enqueue the sync job
54 SyncDataSourceJob.enqueue(instance=datasource, user=request.user)
56 serializer = serializers.DataSourceSerializer(datasource, context={'request': request})
58 return Response(serializer.data)
61class DataFileViewSet(NetBoxReadOnlyModelViewSet):
62 queryset = DataFile.objects.defer('data')
63 serializer_class = serializers.DataFileSerializer
64 filterset_class = filtersets.DataFileFilterSet
67class JobViewSet(NetBoxReadOnlyModelViewSet):
68 """
69 Retrieve a list of job results
70 """
71 queryset = Job.objects.all()
72 serializer_class = serializers.JobSerializer
73 filterset_class = filtersets.JobFilterSet
76class ObjectChangeViewSet(NetBoxReadOnlyModelViewSet):
77 """
78 Retrieve a list of recent changes.
79 """
80 metadata_class = ContentTypeMetadata
81 queryset = ObjectChange.objects.all()
82 serializer_class = serializers.ObjectChangeSerializer
83 filterset_class = filtersets.ObjectChangeFilterSet
85 def get_queryset(self):
86 return super().get_queryset().valid_models()
89class ObjectTypeViewSet(NetBoxReadOnlyModelViewSet):
90 """
91 Read-only list of ObjectTypes.
92 """
93 permission_classes = [IsAuthenticatedOrLoginNotRequired]
94 queryset = ObjectType.objects.order_by('app_label', 'model')
95 serializer_class = serializers.ObjectTypeSerializer
96 filterset_class = filtersets.ObjectTypeFilterSet
98 def initial(self, request, *args, **kwargs):
99 """
100 Override initial() to skip the restrict() call since ObjectType (a ContentType proxy)
101 doesn't use RestrictedQuerySet and is publicly accessible metadata.
102 """
103 # Call GenericViewSet.initial() directly, skipping BaseViewSet.initial()
104 # which would try to call restrict() on the queryset
105 from rest_framework.viewsets import GenericViewSet
106 GenericViewSet.initial(self, request, *args, **kwargs)
109class BaseRQViewSet(viewsets.ViewSet):
110 """
111 Base class for RQ view sets. Provides a list() method. Subclasses must implement get_data().
112 """
113 permission_classes = [IsSuperuser]
114 serializer_class = None
116 def get_data(self):
117 raise NotImplementedError()
119 @extend_schema(responses={200: OpenApiTypes.OBJECT})
120 def list(self, request):
121 data = self.get_data()
122 paginator = LimitOffsetListPagination()
123 data = paginator.paginate_list(data, request)
125 serializer = self.serializer_class(data, many=True, context={'request': request})
126 return paginator.get_paginated_response(serializer.data)
128 def get_serializer(self, *args, **kwargs):
129 """
130 Return the serializer instance that should be used for validating and
131 deserializing input and for serializing output.
132 """
133 serializer_class = self.get_serializer_class()
134 kwargs['context'] = self.get_serializer_context()
135 return serializer_class(*args, **kwargs)
137 def get_serializer_class(self):
138 """
139 Return the class to use for the serializer.
140 """
141 return self.serializer_class
143 def get_serializer_context(self):
144 """
145 Extra context provided to the serializer class.
146 """
147 return {
148 'request': self.request,
149 'format': self.format_kwarg,
150 'view': self,
151 }
154class BackgroundQueueViewSet(BaseRQViewSet):
155 """
156 Retrieve a list of RQ Queues.
157 Note: Queue names are not URL safe, so not returning a detail view.
158 """
159 serializer_class = serializers.BackgroundQueueSerializer
160 lookup_field = 'name'
161 lookup_value_regex = r'[\w.@+-]+'
163 def get_view_name(self):
164 return 'Background Queues'
166 def get_data(self):
167 return get_statistics(run_maintenance_tasks=True)['queues']
169 @extend_schema(
170 operation_id='core_background_queues_retrieve_by_name',
171 parameters=[OpenApiParameter(name='name', type=OpenApiTypes.STR, location=OpenApiParameter.PATH)],
172 responses={200: OpenApiTypes.OBJECT},
173 )
174 def retrieve(self, request, name):
175 data = self.get_data()
176 if not data: 176 ↛ 177line 176 didn't jump to line 177 because the condition on line 176 was never true
177 raise Http404
179 for queue in data:
180 if queue['name'] == name: 180 ↛ 181line 180 didn't jump to line 181 because the condition on line 180 was never true
181 serializer = self.serializer_class(queue, context={'request': request})
182 return Response(serializer.data)
184 raise Http404
187class BackgroundWorkerViewSet(BaseRQViewSet):
188 """
189 Retrieve a list of RQ Workers.
190 """
191 serializer_class = serializers.BackgroundWorkerSerializer
192 lookup_field = 'name'
194 def get_view_name(self):
195 return 'Background Workers'
197 def get_data(self):
198 config = get_queues_list()[0]
199 return Worker.all(get_redis_connection(config['connection_config']))
201 @extend_schema(
202 operation_id='core_background_workers_retrieve_by_name',
203 parameters=[OpenApiParameter(name='name', type=OpenApiTypes.STR, location=OpenApiParameter.PATH)],
204 responses={200: OpenApiTypes.OBJECT},
205 )
206 def retrieve(self, request, name):
207 # all the RQ queues should use the same connection
208 config = get_queues_list()[0]
209 workers = Worker.all(get_redis_connection(config['connection_config']))
210 worker = next((item for item in workers if item.name == name), None)
211 if not worker: 211 ↛ 214line 211 didn't jump to line 214 because the condition on line 211 was always true
212 raise Http404
214 serializer = serializers.BackgroundWorkerSerializer(worker, context={'request': request})
215 return Response(serializer.data)
218class BackgroundTaskViewSet(BaseRQViewSet):
219 """
220 Retrieve a list of RQ Tasks.
221 """
222 serializer_class = serializers.BackgroundTaskSerializer
223 lookup_field = 'id'
225 def get_view_name(self):
226 return 'Background Tasks'
228 def get_data(self):
229 return get_rq_jobs()
231 def get_task_from_id(self, task_id):
232 config = get_queues_list()[0]
233 task = RQ_Job.fetch(task_id, connection=get_redis_connection(config['connection_config']))
234 if not task:
235 raise Http404
237 return task
239 @extend_schema(
240 operation_id='core_background_tasks_retrieve_by_id',
241 parameters=[OpenApiParameter(name='id', type=OpenApiTypes.STR, location=OpenApiParameter.PATH)],
242 responses={200: OpenApiTypes.OBJECT},
243 )
244 def retrieve(self, request, id):
245 """
246 Retrieve the details of the specified RQ Task.
247 """
248 task = self.get_task_from_id(id)
249 serializer = self.serializer_class(task, context={'request': request})
250 return Response(serializer.data)
252 @extend_schema(parameters=[OpenApiParameter(name='id', type=OpenApiTypes.STR, location=OpenApiParameter.PATH)])
253 @action(methods=['POST'], detail=True)
254 def delete(self, request, id):
255 """
256 Delete the specified RQ Task.
257 """
258 delete_rq_job(id)
259 return HttpResponse(status=200)
261 @extend_schema(parameters=[OpenApiParameter(name='id', type=OpenApiTypes.STR, location=OpenApiParameter.PATH)])
262 @action(methods=['POST'], detail=True)
263 def requeue(self, request, id):
264 """
265 Requeues the specified RQ Task.
266 """
267 requeue_rq_job(id)
268 return HttpResponse(status=200)
270 @extend_schema(parameters=[OpenApiParameter(name='id', type=OpenApiTypes.STR, location=OpenApiParameter.PATH)])
271 @action(methods=['POST'], detail=True)
272 def enqueue(self, request, id):
273 """
274 Enqueues the specified RQ Task.
275 """
276 enqueue_rq_job(id)
277 return HttpResponse(status=200)
279 @extend_schema(parameters=[OpenApiParameter(name='id', type=OpenApiTypes.STR, location=OpenApiParameter.PATH)])
280 @action(methods=['POST'], detail=True)
281 def stop(self, request, id):
282 """
283 Stops the specified RQ Task.
284 """
285 stopped_jobs = stop_rq_job(id)
286 if len(stopped_jobs) == 1:
287 return HttpResponse(status=200)
288 return HttpResponse(status=204)