Я пытаюсь создать Rest API, используя DRF. Хотелось получить детали с помощью UniqueId. Я могу использовать PK и получить вывод, но хочу использовать уникальный идентификатор (token_id в моей модели заданий), созданный в поле модели.
Models.py
from django.db import models
from rest_api.util import unique_slug_generator
from django.urls import reverse
# Create your models here.
class Jobs(models.Model):
token_id = models.CharField(max_length=64, unique=True)
name = models.CharField(max_length=100)
url = models.URLField()
environment = models.CharField(max_length=100, null=True)
runtestnow = models.BooleanField(default=False)
def __str__(self):
return self.name
def get_absolute_url(self):
return reverse('token_id', kwargs={'token_id':self.token_id})
class Queue(models.Model):
tokenid = models.ForeignKey(Jobs, on_delete=models.CASCADE)
date = models.DateField(auto_now=True)
def __str__(self):
return self.tokenid
class VM(models.Model):
vm_count = models.IntegerField(default=120)
def __str__(self):
return f"VM Count: {self.vm_count}"
Urls.py
from django.urls import path, include
from . import views
from .views import (RegisterTestMethodView,
RegisterTestMethodViewDetail,
CheckStatusView,
ReleaseTokenView
)
from rest_framework import routers
from rest_framework.authtoken.views import obtain_auth_token
from rest_framework.urlpatterns import format_suffix_patterns
from rest_framework_simplejwt.views import TokenObtainPairView, TokenRefreshView
router = routers.DefaultRouter()
router.register('jobs', views.JobsView)
urlpatterns = [
path('', include(router.urls)),
path('registertestmethod/', RegisterTestMethodView.as_view()),
path('registertestmethod/<int:pk>/',
RegisterTestMethodViewDetail.as_view()),
path('checkstatus/<int:pk>', CheckStatusView.as_view()),
path('checkstatus/<token_id>', CheckStatusView.as_view()),
path('releasetoken/<int:pk>', ReleaseTokenView.as_view()),
]
Serializers.py
from rest_framework import serializers
from .models import Jobs
from django.utils.crypto import get_random_string
class JobsSerializers(serializers.HyperlinkedModelSerializer):
token_id = serializers.CharField(default=get_random_string(length=25))
class Meta:
model = Jobs
fields = ('id', 'name', 'url','runtestnow','token_id')
class CheckStatusSerializers(serializers.HyperlinkedModelSerializer):
class Meta:
model = Jobs
fields = ('id','runtestnow')
class RegisterTestMethodSerializers(serializers.HyperlinkedModelSerializer):
class Meta:
model = Jobs
fields = ('id', 'name', 'url', 'environment', 'runtestnow', 'token_id')
Views.py
from rest_framework import viewsets, permissions, authentication
from .models import Jobs, VM, Queue
from .serializers import (JobsSerializers,
RegisterTestMethodSerializers,
CheckStatusSerializers)
import json
import datetime
import collections
collections.deque()
#3rd time
from rest_framework import generics
from django.http import Http404
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
from rest_framework.authentication import (SessionAuthentication,
BasicAuthentication,
TokenAuthentication)
from rest_framework.permissions import IsAuthenticated
from django.utils.crypto import get_random_string
with open('greet/static/greet/static/config.json', 'r') as
data_config:
data_ready = json.load(data_config)
totalVM = int(data_ready['totalVM'])
max_vm = int(data_ready['max_vm_count'])
grid_name = (data_ready['GridNameForDev'])
min_vm = int(data_ready['min_vm_count'])
class RegisterTestMethodView(APIView):
# authentication_classes = [SessionAuthentication,
TokenAuthentication, BasicAuthentication]
# permission_classes = [IsAuthenticated] # No access (not even
read if not authorized)
def get(self, request):
snippets = Jobs.objects.all()
serializer = RegisterTestMethodSerializers(snippets,
many=True)
return Response(serializer.data)
def post(self, request):
queue = VM.objects.all()
id_token = get_random_string(length=25)
if not queue:
queue = VM(vm_count=totalVM)
queue.save()
else:
for queue_obj in queue:
queue = queue_obj
if queue.vm_count > min_vm:
queue.vm_count -= max_vm
queue.save()
request.data["token_id"] = id_token
request.data["runtestnow"] = True
else:
request.data["token_id"] = id_token
request.data["runtestnow"] = False
serializer = RegisterTestMethodSerializers(data=request.data)
if serializer.is_valid():
serializer.save()
return Response({'TokenId': serializer.data['token_id'],
'RunTestNow': serializer.data['runtestnow'],
'VmCount': queue.vm_count,
'GridName': grid_name, 'Vm_left':
queue.vm_count}, status=status.HTTP_201_CREATED)
return Response(serializer.errors,
status=status.HTTP_400_BAD_REQUEST)
class JobsView(viewsets.ModelViewSet):
queryset = Jobs.objects.all()
serializer_class = JobsSerializers
lookup_field = 'token_id'
class CheckStatusView(APIView):
"""
Retrieve, update or delete a snippet instance.
"""
def get_object(self, pk, token_id):
try:
return Jobs.objects.get(pk=pk)
except Jobs.DoesNotExist:
raise Http404
def get(self, request, token_id):
pk = request.GET.get('pk')
print(pk)
queue = VM.objects.get()
job_list = Jobs.objects.exclude(runtestnow=True)
filtered = Jobs.objects.filter(id=pk)
next_q = job_list.order_by('id').first()
waitlist = 1
return Response(
{"tokenid": token_id, "Runtestnow": False, "VMcount":
queue.vm_count,
'GridName': grid_name, 'waitlist #': waitlist,
'Vm_left':
queue.vm_count}, status=status.HTTP_201_CREATED)
def post(self, request, pk):
queue = VM.objects.get()
vm_count = queue.vm_count
job_list = Jobs.objects.exclude(runtestnow=True)
filtered = Jobs.objects.filter(id=pk)
next_q = job_list.order_by('id').first()
waitlist = int(pk-next_q.id + 1)
if next_q:
print(next_q.id)
if next_q.id == pk and queue.vm_count > min_vm:
queue.vm_count -= max_vm
filtered.update(runtestnow=True)
queue.save()
vm_used = max_vm
else:
filtered.update(runtestnow=False)
queue.save()
vm_used = 0
snippet = self.get_object(pk)
serializer = RegisterTestMethodSerializers(snippet)
return Response({"tokenid": serializer.data["id"],
"Runtestnow": serializer.data['runtestnow'], "VMcount":
vm_used,
'GridName': grid_name, 'waitlist #': waitlist ,
'Vm_left': queue.vm_count},
status=status.HTTP_201_CREATED)
class ReleaseTokenView(APIView):
"""
Retrieve, update or delete a snippet instance.
"""
def get_object(self, pk):
try:
return Jobs.objects.get(pk=pk)
except Jobs.DoesNotExist:
raise Http404
def delete(self, request, pk, format=None):
queue = VM.objects.get()
if not queue:
queue = VM(vm_count=totalVM)
if not self.get_object(pk):
print("Not Method Called...")
return
if queue.vm_count < totalVM :
queue.vm_count += max_vm
queue.save()
elif queue.vm_count + max_vm > totalVM:
queue.vm_count = totalVM
queue.save()
snippet = self.get_object(pk)
snippet.delete()
return Response(data={'Released': True},
status=status.HTTP_204_NO_CONTENT)
Я могу получать информацию, используя, но я хочу пользователя token_id. Я могу сделать это, используя Serializer, как это в рабочих местах. Если я сделаю
localhost/jobs/xJcn8XxF2g9DmmwQwGS0Em754. # --> I get the output but I
# wanna use and I am aware
#that this will return all CRUD methods but how do I apply the
#business logic in Serializers.
localhost/checkstatus/xJcn8XxF2g9DmmwQwGS0Em754 . # --> I wanna
# apply business logic before getting the output. Which
# returns Response related to the PK as well.
Каков наилучший способ сделать это? Должен ли я добавить его на serializer.py (как) или views.py? Буду признателен, если вы предоставите какие-либо полезные документы.