djangorestframework-jwt 用户token认证

2024-11-19 技术教程

1. 项目app01models.pyfrom django.db import modelsfrom django.contrib.auth.models import AbstractUser# Create your models here.class UserProfile(AbstractUser): """ 用户 """ name = models.CharField(max_length=30, null=True, blank=True, verbose_name="姓名") birthday = models.DateField(null=True, blank=True, verbose_name="出生年月") gender = models.CharField(max_length=6, choices=(("male", u"男"), ("female", "女")), default="female", verbose_name="性别") mobile = models.CharField(null=True, blank=True, max_length=11, verbose_name="电话") email = models.EmailField(max_length=100, null=True, blank=True, verbose_name="邮箱") class Meta: verbose_name = "用户" verbose_name_plural = verbose_name def __str__(self): return self.username serializers.py 序列化文件# -*- coding: utf-8 -*-__author__ = 'hyh'from django.contrib.auth import get_user_modelfrom rest_framework import serializersUser = get_user_model()class UserRegSerializer(serializers.ModelSerializer): # code = serializers.CharField(required=True, write_only=True, max_length=4, min_length=4,label="验证码", # error_messages={ # "blank": "请输入验证码", # "required": "请输入验证码", # "max_length": "验证码格式错误", # "min_length": "验证码格式错误" # }, # help_text="验证码") username = serializers.CharField(label="用户名", help_text="用户名", required=True, allow_blank=False, validators=[UniqueValidator(queryset=User.objects.all(), message="用户已经存在")]) password = serializers.CharField( style={'input_type': 'password'},help_text="密码", label="密码", write_only=True, ) def create(self, validated_data): user = super(UserRegSerializer, self).create(validated_data=validated_data) user.set_password(validated_data["password"]) user.save() return user # def validate_code(self, code): # # try: # # verify_records = VerifyCode.objects.get(mobile=self.initial_data["username"], code=code) # # except VerifyCode.DoesNotExist as e: # # pass # # except VerifyCode.MultipleObjectsReturned as e: # # pass # verify_records = VerifyCode.objects.filter(mobile=self.initial_data["username"]).order_by("-add_time") # if verify_records: # last_record = verify_records[0] # # five_mintes_ago = datetime.now() - timedelta(hours=0, minutes=5, seconds=0) # if five_mintes_ago > last_record.add_time: # raise serializers.ValidationError("验证码过期") # # if last_record.code != code: # raise serializers.ValidationError("验证码错误") # # else: # raise serializers.ValidationError("验证码错误") def validate(self, attrs): attrs["mobile"] = attrs["username"] #del attrs["code"] return attrs class Meta: model = User fields = "__all__"2.安装djangorestframework-jwtpip install djangorestframework-jwt配置settings文件添加如下配置import datetimeREST_FRAMEWORK = { 'DEFAULT_AUTHENTICATION_CLASSES': ( #'rest_framework_jwt.authentication.JSONWebTokenAuthentication', # 全局设置 'rest_framework.authentication.BasicAuthentication', 'rest_framework.authentication.SessionAuthentication', ),}JWT_AUTH = { 'JWT_EXPIRATION_DELTA': datetime.timedelta(days=7), 'JWT_AUTH_HEADER_PREFIX': 'JWT',}AUTH_USER_MODEL = 'app01.UserProfile' # 自定义用户表#AUTHENTICATION_BACKENDS = (# # 将backends添加进setting# 'app01.views.CustomBackend',#) 自定义认证不可用# redis缓存CACHES = { "default": { "BACKEND": "django_redis.cache.RedisCache", "LOCATION": "redis://127.0.0.1:6379", "OPTIONS": { "CLIENT_CLASS": "django_redis.client.DefaultClient", "PASSWORD": "123456", } }}view.pyfrom django.shortcuts import renderfrom django.contrib.auth.backends import ModelBackendfrom rest_framework import viewsetsfrom rest_framework import mixinsfrom rest_framework import authenticationfrom rest_framework_jwt.authentication import JSONWebTokenAuthenticationfrom rest_framework_jwt.serializers import jwt_encode_handler, jwt_payload_handlerfrom django.contrib.auth import get_user_modelfrom .serializers import UserRegSerializerfrom django.db.models import Qfrom rest_framework import statusfrom rest_framework.views import APIViewfrom rest_framework.response import Responsefrom .models import UserProfilefrom django.http import JsonResponsefrom rest_framework.permissions import IsAuthenticatedfrom rest_framework import permissionsfrom app01.utils.permissions import IsOwnerOrReadOnlyfrom rest_framework_extensions.cache.mixins import CacheResponseMixin #pip3 install drf-extensions # 使用缓存ListCacheResponseMixin,RetrieveCacheResponseMixin# Create your views here.User = get_user_model()#class CustomBackend(ModelBackend):# """# 自定义用户验证# """# def authenticate(self, username=None, password=None, **kwargs):# try:# user = UserProfile.objects.get(Q(username=username)|Q(mobile=username))# if user.check_password(password):# return user# except Exception as e:# return Noneclass UserViewset(CacheResponseMixin,mixins.CreateModelMixin, mixins.UpdateModelMixin, mixins.RetrieveModelMixin,mixins.ListModelMixin, viewsets.GenericViewSet): serializer_class = UserRegSerializer queryset = UserProfile.objects.all() #permission_classes = (IsAuthenticated, IsOwnerOrReadOnly) # 权限和认证必须同时使用，否则认证不生效 authentication_classes = (JSONWebTokenAuthentication,authentication.SessionAuthentication) def get_serializer_class(self): if self.action == "retrieve": return UserRegSerializer elif self.action == "create": return UserRegSerializer return UserProfileSerializer # permission_classes = (permissions.IsAuthenticated, ) def get_permissions(self): if self.action == "retrieve": return [permissions.IsAuthenticated()] elif self.action == "list": return [permissions.IsAuthenticated()] #list也需要认证 elif self.action == "create": return [] return [] def create(self, request, *args, **kwargs): serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) user = self.perform_create(serializer) re_dict = serializer.data payload = jwt_payload_handler(user) re_dict["token"] = jwt_encode_handler(payload) re_dict["name"] = user.name if user.name else user.username headers = self.get_success_headers(serializer.data) return Response(re_dict, status=status.HTTP_201_CREATED, headers=headers) def get_object(self): return self.request.user def perform_create(self, serializer): return serializer.save()class Index(CacheResponseMixin,APIView): permission_classes = (IsAuthenticated, IsOwnerOrReadOnly) authentication_classes = (JSONWebTokenAuthentication,) def get(self,request): return JsonResponse({"index":"ok"}) 3.urls.pyfrom django.contrib import adminfrom django.urls import pathfrom django.conf.urls import url,includefrom rest_framework.routers import SimpleRouter,DefaultRouterfrom rest_framework.documentation import include_docs_urlsfrom app01.views import UserViewset,Indexfrom rest_framework_jwt.views import obtain_jwt_tokenrouter = SimpleRouter()router.register('user',UserViewset,base_name='useruinfo')urlpatterns = [ path('admin/', admin.site.urls), url(r'^api-auth/', include('rest_framework.urls', namespace='rest_framework')), url(r'docs/', include_docs_urls(title="user信息")), url(r'^login/',obtain_jwt_token), url(r'^index/',Index.as_view()),]urlpatterns += router.urls 4. 在app01下创建utils目录，存放permissions.py文件 permissions.py文件内容 # -*- coding: utf-8 -*-__author__ = 'hyh'from rest_framework import permissionsclass IsOwnerOrReadOnly(permissions.BasePermission): """ Object-level permission to only allow owners of an object to edit it. Assumes the model instance has an `owner` attribute. """ def has_object_permission(self, request, view, obj): # Read permissions are allowed to any request, # so we'll always allow GET, HEAD or OPTIONS requests. if request.method in permissions.SAFE_METHODS: return True # Instance must have an attribute named `owner`. return obj.user == request.user 4.测试获取token

复制token

返回结果成功