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

7848 switch to APIView

Browse Source
This commit is contained in:
Arthur Hanson 2024-11-06 12:02:38 -08:00
parent 21a61dcb66
commit b9f4f93ca0
2 changed files with 49 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,6 +1,9 @@
from django.urls import path, include
from netbox.api.routers import NetBoxRouter
from . import views
app_name = 'core-api'
router = NetBoxRouter()
router.APIRootView = views.CoreRootView
@ -9,15 +12,32 @@ router.register('data-sources', views.DataSourceViewSet)
router.register('data-files', views.DataFileViewSet)
router.register('jobs', views.JobViewSet)
router.register('object-changes', views.ObjectChangeViewSet)
router.register('background-workers', views.BackgroundWorkerViewSet, basename='BackgroundWorkers')
router.register('background-queues', views.BackgroundQueueViewSet, basename='BackgroundQueues')
router.register('background-tasks/(?P<queue_name>[\w-]+)', views.BackgroundTaskViewSet, basename='BackgroundTasks')
router.register('background-tasks/(?P<queue_name>[\w-]+)/deferred', views.BackgroundTaskDeferredViewSet, basename='BackgroundTaskDeferred')
router.register('background-tasks/(?P<queue_name>[\w-]+)/failed', views.BackgroundTaskFailedViewSet, basename='BackgroundTaskFailed')
router.register('background-tasks/(?P<queue_name>[\w-]+)/finished', views.BackgroundTaskFinishedViewSet, basename='BackgroundTaskFinished')
router.register('background-tasks/(?P<queue_name>[\w-]+)/started', views.BackgroundTaskStartedViewSet, basename='BackgroundTaskStarted')
router.register('background-tasks/(?P<queue_name>[\w-]+)/queued', views.BackgroundTaskQueuedViewSet, basename='BackgroundTaskQueued')
router.register('background-tasks/(?P<queue_name>[\w-]+)/workers', views.BackgroundTaskWorkerViewSet, basename='BackgroundTaskWorkers')
app_name = 'core-api'
urlpatterns = router.urls
# Background Tasks
"""
router.register('background-queues/', views.BackgroundQueueListView.as_view(), name='background_queue_list'),
router.register('background-queues/<int:queue_index>/<str:status>/', views.BackgroundTaskListView.as_view(), name='background_task_list'),
router.register('background-tasks/<str:job_id>/', views.BackgroundTaskView.as_view(), name='background_task'),
router.register('background-tasks/<str:job_id>/delete/', views.BackgroundTaskDeleteView.as_view(), name='background_task_delete'),
router.register('background-tasks/<str:job_id>/requeue/', views.BackgroundTaskRequeueView.as_view(), name='background_task_requeue'),
router.register('background-tasks/<str:job_id>/enqueue/', views.BackgroundTaskEnqueueView.as_view(), name='background_task_enqueue'),
router.register('background-tasks/<str:job_id>/stop/', views.BackgroundTaskStopView.as_view(), name='background_task_stop'),
router.register('background-workers/<int:queue_index>/', views.WorkerListView.as_view(), name='worker_list'),
router.register('background-workers/<str:key>/', views.WorkerView.as_view(), name='worker'),
"""
urlpatterns = (
path('background-queues/', views.QueueListView.as_view()),
path('background-workers/', views.WorkerListView.as_view()),
# path('background-workers/<str:key>/', views.WorkerView.as_view()),
path('background-tasks/<str:queue_name>/', views.TaskListView.as_view()),
path('background-tasks/<str:queue_name>/deferred/', views.DeferredTaskListView.as_view()),
path('background-tasks/<str:queue_name>/failed/', views.FailedTaskListView.as_view()),
path('background-tasks/<str:queue_name>/finished/', views.FinishedTaskListView.as_view()),
path('background-tasks/<str:queue_name>/started/', views.StartedTaskListView.as_view()),
path('background-tasks/<str:queue_name>/queued/', views.QueuedTaskListView.as_view()),
# path('background-tasks/<str:queue_name>/workers', views.BackgroundWorkerListView.as_view()),
path('', include(router.urls)),
)

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

@ -6,7 +6,8 @@ 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 rest_framework.views import APIView
from rest_framework.viewsets import ReadOnlyModelViewSet
from core import filtersets
from core.choices import DataSourceStatusChoices
@ -20,7 +21,6 @@ from netbox.api.metadata import ContentTypeMetadata
from netbox.api.viewsets import NetBoxModelViewSet, NetBoxReadOnlyModelViewSet
from rest_framework.permissions import IsAdminUser
from rq.worker import Worker
from rq.worker_registration import clean_worker_registry
from . import serializers
@ -82,21 +82,20 @@ class ObjectChangeViewSet(ReadOnlyModelViewSet):
filterset_class = filtersets.ObjectChangeFilterSet
class BackgroundQueueViewSet(ViewSet):
serializer_class = serializers.BackgroundQueueSerializer
class QueueListView(APIView):
permission_classes = [IsAdminUser]
def get_view_name(self):
return "BackgroundQueueViewSet"
return "Background Queues"
@extend_schema(responses={200: OpenApiTypes.OBJECT})
def list(self, request):
def get(self, request, format=None):
data = get_statistics(run_maintenance_tasks=True)["queues"]
serializer = serializers.BackgroundQueueSerializer(data, many=True)
return Response(serializer.data)
class BackgroundWorkerViewSet(ViewSet):
class WorkerListView(APIView):
serializer_class = serializers.BackgroundQueueSerializer
permission_classes = [IsAdminUser]
@ -104,7 +103,7 @@ class BackgroundWorkerViewSet(ViewSet):
return "Background Workers"
@extend_schema(responses={200: OpenApiTypes.OBJECT})
def list(self, request):
def get(self, request, format=None):
# all the RQ queues should use the same connection
config = QUEUES_LIST[0]
workers = Worker.all(get_redis_connection(config['connection_config']))
@ -112,15 +111,15 @@ class BackgroundWorkerViewSet(ViewSet):
return Response(serializer.data)
class BackgroundTaskViewSet(ViewSet):
class TaskListView(APIView):
serializer_class = serializers.BackgroundTaskSerializer
permission_classes = [IsAdminUser]
def get_view_name(self):
return "BackgroundTaskViewSet"
return "Background Tasks"
@extend_schema(responses={200: OpenApiTypes.OBJECT})
def list(self, request, queue_name):
def get(self, request, format=None):
"""
Return the UserConfig for the currently authenticated User.
"""
@ -130,7 +129,7 @@ class BackgroundTaskViewSet(ViewSet):
return Response(serializer.data)
class BaseBackgroundTaskViewSet(ViewSet):
class BaseTaskView(APIView):
serializer_class = serializers.BackgroundTaskSerializer
permission_classes = [IsAdminUser]
registry = None
@ -139,46 +138,46 @@ class BaseBackgroundTaskViewSet(ViewSet):
return "Background Tasks"
@extend_schema(responses={200: OpenApiTypes.OBJECT})
def list(self, request, queue_name):
def get(self, request, queue_name, format=None):
queue = get_queue(queue_name)
data = get_rq_jobs_from_status(queue, self.registry)
serializer = serializers.BackgroundTaskSerializer(data, many=True)
return Response(serializer.data)
class BackgroundTaskDeferredViewSet(BaseBackgroundTaskViewSet):
class DeferredTaskListView(BaseTaskView):
registry = "deferred"
def get_view_name(self):
return "Deferred Tasks"
class BackgroundTaskFailedViewSet(BaseBackgroundTaskViewSet):
class FailedTaskListView(BaseTaskView):
registry = "failed"
def get_view_name(self):
return "Failed Tasks"
class BackgroundTaskFinishedViewSet(BaseBackgroundTaskViewSet):
class FinishedTaskListView(BaseTaskView):
registry = "finished"
def get_view_name(self):
return "Finished Tasks"
class BackgroundTaskStartedViewSet(BaseBackgroundTaskViewSet):
class StartedTaskListView(BaseTaskView):
registry = "started"
def get_view_name(self):
return "Started Tasks"
class BackgroundTaskQueuedViewSet(BaseBackgroundTaskViewSet):
class QueuedTaskListView(BaseTaskView):
registry = "queued"
@extend_schema(responses={200: OpenApiTypes.OBJECT})
def list(self, request, queue_name):
def get(self, request, format=None):
"""
Return the UserConfig for the currently authenticated User.
"""
@ -186,20 +185,3 @@ class BackgroundTaskQueuedViewSet(BaseBackgroundTaskViewSet):
data = queue.get_jobs()
serializer = serializers.BackgroundTaskSerializer(data, many=True)
return Response(serializer.data)
class BackgroundTaskWorkerViewSet(ViewSet):
serializer_class = serializers.BackgroundQueueSerializer
permission_classes = [IsAdminUser]
def get_view_name(self):
return "Background Workers"
@extend_schema(responses={200: OpenApiTypes.OBJECT})
def list(self, request, queue_name):
queue = get_queue(queue_name)
clean_worker_registry(queue)
all_workers = Worker.all(queue.connection)
workers = [worker for worker in all_workers if queue.name in worker.queue_names()]
serializer = serializers.BackgroundWorkerSerializer(workers, many=True)
return Response(serializer.data)