???

Posted 2022-12-26 xzqpy

1 . 

url

from rest_framework_jwt.views import  obtain_jwt_token

urlpatterns = [
    path(‘admin/‘, admin.site.urls),
    path(‘login/‘,obtain_jwt_token),
]

SETTING

REST_FRAMEWORK = 
    ‘JWT_EXPIRATION_DELTA‘: datetime.timedelta(seconds=60*60),
　　

 

2.MODEL

from django.db import models
import uuid
# Create your models here.


class Emp(models.Model):
  
    uuid = models.UUIDField(primary_key=True, default=uuid.uuid4)
    name = models.CharField(null = False, max_length =30)
    empno = models.BigIntegerField(unique = True, db_index=True,null = False)
    group = models.ManyToManyField(to="Org")
    
    def save(self, force_insert=False, force_update=False, using=None,
             update_fields=None):
        if int(self.empno) > 50000:
            raise ValueError(‘工号不可大于50000‘)
        return super().save(force_insert=False, force_update=False, using=None, update_fields=None)


class Org(models.Model):
    
    name = models.CharField(unique=True, db_index=True, max_length=50)
    positions = ((1, ‘开发‘), (2, ‘运维‘), (3, ‘产品‘), (4, ‘测试‘))
    type = models.IntegerField(choices=positions) 

VIEW

from app02 import models
from app02 import serializers
from app02 import authenticationclasseses
from rest_framework.response import Response
from rest_framework.generics import ListAPIView 
from rest_framework.pagination import PageNumberPagination
from rest_framework_jwt.authentication import JSONWebTokenAuthentication
from django_filters.rest_framework import DjangoFilterBackend


class EmployeeView(JSONWebTokenAuthentication, ListAPIView):
    
    authentication_classes = [
                              # authenticationclasseses.MyAuth,
                              authenticationclasseses.MyJSONWebTokenAuthentication,
                              ]
    queryset = models.Emp.objects.all()
    serializer_class = serializers.EmployeeSerializers
    filter_backends = [DjangoFilterBackend]  # 排序过滤
    filter_fields = [‘uuid‘, ‘name‘, ‘empno‘]  # 可以过滤的字段
    
    def list(self, request, *args, **kwargs):
        queryset = self.filter_queryset(self.get_queryset())
        page = self.paginate_queryset(queryset)
        if page is not None:
            serializer = self.get_serializer(page, many=True)
            return self.get_paginated_response(serializer.data)

        page = PageNumberPagination()
        # 在数据库中获取分页的数据
        queryset = page.paginate_queryset(self.queryset, request, view=self)
        serializer = self.get_serializer(queryset, many=True)
        return Response(serializer.data)


class EmployeedetailView(ListAPIView):
    
    queryset = models.Emp.objects.all()
    serializer_class = serializers.EmployeeSerializers
    
    def list(self, request, *args, **kwargs):
        self.queryset = self.queryset.filter(empno=kwargs["empno"])
        response = super().list(request, *args, **kwargs)
        return response

authenticationclasseses.py

import jwt
from rest_framework import exceptions
from rest_framework.exceptions import AuthenticationFailed
from django.utils.translation import ugettext as _
from rest_framework_jwt.settings import api_settings
from django.contrib.auth import get_user_model
from rest_framework_jwt.authentication import JSONWebTokenAuthentication
jwt_decode_handler = api_settings.JWT_DECODE_HANDLER
jwt_get_username_from_payload = api_settings.JWT_PAYLOAD_GET_USERNAME_HANDLER


class MyAuth(object):
    def authenticate(self, request):
        username = request.GET.get("username")
        password = request.GET.get("password")
        user = get_user_model().objects.filter(username=username)
        try:
            is_bingo = user.first().check_password(password)
        except Exception as f:
            print(f)
            raise AuthenticationFailed("账号不存在")
        if is_bingo:
            return username, password
        raise AuthenticationFailed("密码不正确")
    
    def authenticate_header(self, request):
        pass


class MyJSONWebTokenAuthentication(JSONWebTokenAuthentication):

    def authenticate(self, request):
        jwt_value = self.get_jwt_value(request)
        if jwt_value is None:
            raise exceptions.AuthenticationFailed("必须传入token")

        try:
            payload = jwt_decode_handler(jwt_value)
        except jwt.ExpiredSignature:
            msg = _(‘Signature has expired.‘)
            raise exceptions.AuthenticationFailed(msg)
        except jwt.DecodeError:
            msg = _(‘Error decoding signature.‘)
            raise exceptions.AuthenticationFailed(msg)
        except jwt.InvalidTokenError:
            raise exceptions.AuthenticationFailed()
        user = self.authenticate_credentials(payload)
        return (user, jwt_value) 

　　

 

serializers.py

from rest_framework import serializers
from app02 import models
from rest_framework.pagination import PageNumberPagination


class OrgSerializers(serializers.ModelSerializer):
    class Meta:
        model = models.Org
        fields = ["name"]
        depth = 1


class EmployeeSerializers(serializers.ModelSerializer):
    group = OrgSerializers(many=True)
    
    class Meta:
        model = models.Emp
        fields = ["empno", "name", "group"]


class Pager(PageNumberPagination):
    
    page_size = 2
    page_query_param = ‘page‘
    # 定制传参
    page_size_query_param = ‘size‘
    # 最大一页的数据
    max_page_size = 5　　

　

SETING

REST_FRAMEWORK = 
    ‘JWT_EXPIRATION_DELTA‘: datetime.timedelta(seconds=60*60),
    ‘PAGE_SIZE‘:2,

 

URL

from django.contrib import admin
from django.urls import path,re_path
from rest_framework_jwt.views import  obtain_jwt_token
from rest_framework.generics import  ListCreateAPIView
from app02 import  views
urlpatterns = [
    path(‘admin/‘, admin.site.urls),
    path(‘login/‘,obtain_jwt_token),
    re_path(r‘employee/(?P<empno>\d+)/‘, views.EmployeedetailView.as_view()),
    path(r‘employee/‘, views.EmployeeView.as_view()),
]

ADMIN

admin.site.register(models.Org)
admin.site.register(models.Emp)