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 cleanup

Browse Source
This commit is contained in:
Arthur Hanson 2024-11-19 11:00:04 -08:00
parent 47beb7f160
commit 10778adc47
2 changed files with 4 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
netbox/core/api/serializers_/tasks.py
View File

@ -1,5 +1,4 @@
from rest_framework import serializers
from rest_framework.reverse import reverse
__all__ = (
'BackgroundTaskSerializer',
@ -38,10 +37,10 @@ class BackgroundTaskSerializer(serializers.Serializer):
is_scheduled = serializers.BooleanField()
is_stopped = serializers.BooleanField()
def get_position(self, obj):
def get_position(self, obj) -> int:
return obj.get_position()
def get_status(self, obj):
def get_status(self, obj) -> str:
return obj.get_status()
@ -60,10 +59,6 @@ class BackgroundQueueSerializer(serializers.Serializer):
failed_jobs = serializers.IntegerField()
scheduled_jobs = serializers.IntegerField()
def get_url(self, obj):
request = self.context.get('request')
return reverse('core-api:rqqueue-detail', kwargs={'name': obj['name']}, request=request)
class BackgroundWorkerSerializer(serializers.Serializer):
name = serializers.CharField()

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

@ -13,8 +13,8 @@ from core import filtersets
from core.choices import DataSourceStatusChoices
from core.jobs import SyncDataSourceJob
from core.models import *
from core.utils import delete_rq_job, enqueue_rq_job, get_rq_jobs, get_rq_jobs_from_status, requeue_rq_job, stop_rq_job
from django_rq.queues import get_queue, get_redis_connection
from core.utils import delete_rq_job, enqueue_rq_job, get_rq_jobs, requeue_rq_job, stop_rq_job
from django_rq.queues import get_redis_connection
from django_rq.utils import get_statistics
from django_rq.settings import QUEUES_LIST
from netbox.api.metadata import ContentTypeMetadata
@ -155,23 +155,6 @@ class TaskViewSet(BaseRQListView):
def get_view_name(self):
return "Background Tasks"
def get_response(self, request, queue_name, status=None):
try:
queue = get_queue(queue_name)
except KeyError:
raise Http404
if status:
data = get_rq_jobs_from_status(queue, status)
else:
data = queue.get_jobs()
paginator = LimitOffsetListPagination()
data = paginator.paginate_list(data, request)
serializer = serializers.BackgroundTaskSerializer(data, many=True, context={'request': request})
return Response(serializer.data)
def get_data(self):
return get_rq_jobs()