在 Django Rest 框架 URL 中通过唯一 ID 但不是 PK 获取详细信息

Posted 2023-02-24

【中文标题】在 Django Rest 框架 URL 中通过唯一 ID 但不是 PK 获取详细信息

【英文标题】：Get detail by Unique ID but not PK in Django Rest Framework URLs

【发布时间】：2020-05-27 06:58:01

【问题描述】：

我正在尝试使用 DRF 创建 Rest API。想通过使用 UniqueId 来获取详细信息。我可以使用 PK 并获取输出，但想使用在模型字段中创建的唯一 ID（我的工作模型中的 token_id）。

模型.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"

网址.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()),

]

序列化器.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 获取信息。我可以像在工作中那样使用序列化器来做到这一点。 如果我这样做了

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(how) 或 views.py 中？ 如果您提供任何有用的文件，我将不胜感激。

【问题讨论】：

我认为你包含了错误的观点。您应该包括作业视图和检查状态视图。

我是堆栈溢出的新手，要缩进所有内容并确保所有内容都完美无缺是非常困难的。我已经在这里添加了我的全部观点，请您看一下。

ishak 的答案似乎是正确的。或者至少在正确的方向上。

感谢@schillingt 的帮助。真的很感激

【参考方案1】：

您应该在序列化程序和视图集中将lookup_field 设置为token_id。

这里是答案Django Rest Framework: Access item detail by slug instead of ID

【参考方案2】：

实际上，我通过一些研究能够做到这一点。似乎我必须在 URL 中传递一个唯一 ID（token_id），并在 views.py 上使用相同的唯一 ID（token_id）进行查询。我知道有一个模型视图集可以毫不费力地完成它，正如 Ishak 所提到的，但我想使用 APIView，除此之外我还想添加一些业务逻辑。我可能需要对如何向模型视图集添加逻辑进行更多研究。这是我的解决方案。

Views.py

    def get(self, request, token_id):

        get_job = Jobs.objects.get(token_id=token_id)
        pk = get_job.id


        job_list = Jobs.objects.exclude(runtestnow=True)
        next_q = job_list.order_by('id').first()

        queue = VM.objects.get()

        waitlist = int(pk) - int(next_q.id) 

        if waitlist == 1:
            waitlist = 'You are next on the queue. :)'

        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)

网址.py

path('checkstatus/<token_id>', CheckStatusView.as_view()),

我们总是可以使用 slug 字段，但我真的希望 token_id 作为输入。到目前为止，这对我来说应该可以正常工作。 可能还有其他方法。随意分享。