NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-07-29 03:46:25 -06:00
Code Issues Actions 8 Packages Projects Releases Wiki Activity

7848 more viewsets

Browse Source
This commit is contained in:
Arthur Hanson 2024-11-18 16:57:29 -08:00
parent 8ffec29403
commit 46a39a47be
3 changed files with 23 additions and 101 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
netbox/core/api/urls.py
View File

@ -1,5 +1,3 @@
from django.urls import path, include
from netbox.api.routers import NetBoxRouter
from . import views
@ -15,13 +13,6 @@ router.register('object-changes', views.ObjectChangeViewSet)
router.register('background-queues', views.QueueViewSet, basename='rqqueue')
router.register('background-workers', views.WorkerViewSet, basename='rqworker')
router.register('background-tasks/(?P<queue_name>.+)/', views.TaskViewSet, basename='rqtask')
router.register('background-task', views.TaskDetailViewSet, basename='rqtaskdetail')
urlpatterns = (
path('background-task/<str:task_id>/', views.TaskDetailView.as_view(), name="background_task_detail"),
path('background-task/<str:task_id>/delete/', views.TaskDeleteView.as_view(), name="background_task_delete"),
path('background-task/<str:task_id>/requeue/', views.TaskRequeueView.as_view(), name="background_task_requeue"),
path('background-task/<str:task_id>/enqueue/', views.TaskEnqueueView.as_view(), name="background_task_enqueue"),
path('background-task/<str:task_id>/stop/', views.TaskStopView.as_view(), name="background_task_stop"),
path('', include(router.urls)),
)
urlpatterns = router.urls

101
netbox/core/api/views.py
View File

@ -7,7 +7,6 @@ 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.views import APIView
from rest_framework.viewsets import ReadOnlyModelViewSet
from core import filtersets
@ -171,30 +170,12 @@ class WorkerViewSet(BaseRQListView):
return Response(serializer.data)
class TaskListView(APIView):
class TaskDetailViewSet(viewsets.ViewSet):
"""
Retrieve a list of RQ Tasks in the specified Queue.
Retrieve the details of the specified RQ Task.
"""
permission_classes = [IsAdminUser]
def get_view_name(self):
return "Background Tasks"
@extend_schema(responses={200: OpenApiTypes.OBJECT})
def get(self, request, queue_name, format=None):
try:
queue = get_queue(queue_name)
except KeyError:
raise Http404
data = queue.get_jobs()
serializer = serializers.BackgroundTaskSerializer(data, many=True, context={'request': request})
return Response(serializer.data)
class BaseTaskView(APIView):
permission_classes = [IsAdminUser]
def get_view_name(self):
return "Background Task"
@ -206,80 +187,30 @@ class BaseTaskView(APIView):
return task
class TaskDetailView(BaseTaskView):
"""
Retrieve the details of the specified RQ Task.
"""
permission_classes = [IsAdminUser]
def get_view_name(self):
return "Background Task"
@extend_schema(responses={200: OpenApiTypes.OBJECT})
def get(self, request, task_id, format=None):
task = self.get_task_from_id(task_id)
def retrieve(self, request, pk):
task = self.get_task_from_id(pk)
serializer = serializers.BackgroundTaskSerializer(task, context={'request': request})
return Response(serializer.data)
class TaskDeleteView(APIView):
"""
Deletes the specified RQ Task.
"""
permission_classes = [IsAdminUser]
def get_view_name(self):
return "Background Task"
@extend_schema(responses={200: OpenApiTypes.OBJECT})
def post(self, request, task_id, format=None):
delete_rq_job(task_id)
@action(methods=["POST"], detail=True)
def delete(self, request, pk):
delete_rq_job(pk)
return HttpResponse(status=200)
class TaskRequeueView(APIView):
"""
Requeues the specified RQ Task.
"""
permission_classes = [IsAdminUser]
def get_view_name(self):
return "Background Task"
@extend_schema(responses={200: OpenApiTypes.OBJECT})
def post(self, request, task_id, format=None):
requeue_rq_job(task_id)
@action(methods=["POST"], detail=True)
def requeue(self, request, pk):
requeue_rq_job(pk)
return HttpResponse(status=200)
class TaskEnqueueView(APIView):
"""
Enqueues the specified RQ Task.
"""
permission_classes = [IsAdminUser]
def get_view_name(self):
return "Background Task"
@extend_schema(responses={200: OpenApiTypes.OBJECT})
def post(self, request, task_id, format=None):
enqueue_rq_job(task_id)
@action(methods=["POST"], detail=True)
def enqueue(self, request, pk):
enqueue_rq_job(pk)
return HttpResponse(status=200)
class TaskStopView(APIView):
"""
Stops the specified RQ Task.
"""
permission_classes = [IsAdminUser]
def get_view_name(self):
return "Background Task"
@extend_schema(responses={200: OpenApiTypes.OBJECT})
def post(self, request, task_id, format=None):
stopped_jobs = stop_rq_job(task_id)
@action(methods=["POST"], detail=True)
def stop(self, request, pk):
stopped_jobs = stop_rq_job(pk)
if len(stopped_jobs) == 1:
return HttpResponse(status=200)
else:

10
netbox/core/tests/test_api.py
View File

@ -193,7 +193,7 @@ class BackgroundTaskTestCase(TestCase):
queue = get_queue('default')
job = queue.enqueue(self.dummy_job_default)
response = self.client.get(reverse('core-api:background_task_detail', args=[job.id]), **self.header)
response = self.client.get(reverse('core-api:rqtaskdetail-detail', args=[job.id]), **self.header)
self.assertEqual(response.status_code, 200)
self.assertIn(str(job.id), str(response.content))
self.assertIn('origin', str(response.content))
@ -204,7 +204,7 @@ class BackgroundTaskTestCase(TestCase):
queue = get_queue('default')
job = queue.enqueue(self.dummy_job_default)
response = self.client.post(reverse('core-api:background_task_delete', args=[job.id]), **self.header)
response = self.client.post(reverse('core-api:rqtaskdetail-delete', args=[job.id]), **self.header)
self.assertEqual(response.status_code, 200)
self.assertFalse(RQ_Job.exists(job.id, connection=queue.connection))
queue = get_queue('default')
@ -220,7 +220,7 @@ class BackgroundTaskTestCase(TestCase):
self.assertTrue(job.is_failed)
# Re-enqueue the failed job and check that its status has been reset
response = self.client.post(reverse('core-api:background_task_requeue', args=[job.id]), **self.header)
response = self.client.post(reverse('core-api:rqtaskdetail-requeue', args=[job.id]), **self.header)
self.assertEqual(response.status_code, 200)
job = RQ_Job.fetch(job.id, queue.connection)
self.assertFalse(job.is_failed)
@ -240,7 +240,7 @@ class BackgroundTaskTestCase(TestCase):
self.assertIsNone(job.enqueued_at)
# Force-enqueue the deferred job
response = self.client.post(reverse('core-api:background_task_enqueue', args=[job.id]), **self.header)
response = self.client.post(reverse('core-api:rqtaskdetail-enqueue', args=[job.id]), **self.header)
self.assertEqual(response.status_code, 200)
# Check that job's status is updated correctly
@ -256,7 +256,7 @@ class BackgroundTaskTestCase(TestCase):
worker.prepare_job_execution(job)
self.assertEqual(job.get_status(), JobStatus.STARTED)
response = self.client.post(reverse('core-api:background_task_stop', args=[job.id]), **self.header)
response = self.client.post(reverse('core-api:rqtaskdetail-stop', args=[job.id]), **self.header)
self.assertEqual(response.status_code, 200)
worker.monitor_work_horse(job, queue) # Sets the job as Failed and removes from Started
started_job_registry = StartedJobRegistry(queue.name, connection=queue.connection)