京西商城——基于DRF认证和JWT实现用户登录及Token认证

第一步：登陆后生成Token

在用户出入的信息验证成功的时候，服务端在响应中加入Token，以便之后用户再次访问时携带Token用来认证

user/views.py
class UserLoginView(GenericAPIView):
    serializer_class = UserSerializer
    queryset = User.objects

    def post(self, request):
        data = request.data
        # print(data)
        email = data['email']
        # print(email)
        user = User.objects.filter(email=email).first()
        # print(user)
        if not user:
            return ResponseMessage.UserResponse.other('用户名不存在或者密码错误')
        else:
            password = data['password']
            form_password = get_md5(password)
            db_password = user.password
            if form_password != db_password:
                return ResponseMessage.UserResponse.other('用户名不存在或者密码错误')
            else:
                token_payload = {
                    'email': user.email,
                }
                token = create_token(token_payload, timeout=1)
                result = {
                    'token': token,
                    'username': user.name,
                    'status': '登陆成功'
                }
                return ResponseMessage.UserResponse.success(result)

utils/jwt.py
import datetime
import jwt
from django.conf.global_settings import SECRET_KEY

# 生成Token
def create_token(payload, timeout):
    header = {
        "alg": "HS256",
        "typ": "JWT"
    }

    payload = payload
    payload["exp"] = datetime.datetime.utcnow() + datetime.timedelta(minutes=timeout)

    result = jwt.encode(headers=header, payload=payload, key=SECRET_KEY, algorithm='HS256')
    return result

可以看到此时在登陆后的返回信息中包含一个Token

第二步：用户请求携带Token，views进行验证

我们要先编写验证Token的代码，定义一个Token的身份验证类

utils/jwt.py
# 验证Token
def get_payload(token):
    result = {"status": False, "payload": None, "error": None}
    try :
        payload = jwt.decode(jwt=token, key=SECRET_KEY, algorithms=['HS256'])
        result["status"] = True
        result['payload'] = payload
    except jwt.exceptions.ExpiredSignatureError:
        result['error'] = 'Token已经过期了'
    except jwt.exceptions.InvalidSignatureError:
        result['error'] = 'Token不合法'
    except jwt.exceptions.DecodeError:
        result['error'] = 'Token不能通过验证'
    return result

# JWT，URL参数传递Token的身份验证类
class JWTURLParamAuthentication(BaseAuthentication):
    def authenticate(self, request):
        token = request.GET.get('token')
        token_data = get_payload(token)
        print(token_data)
        return token_data, token
        # 这里必选按照
        # {'status': True, 'payload': {'email': '006@QQ.com', 'exp': 1712306747}, 'error': None}
        # eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJlbWFpbCI6IjAwNkBRUS5jb20iLCJleHAiOjE3MTIzMDY3NDd9.uq3kYZaksxVNocq6rzYk2Fb8xb6e4PQ84fV6GKed3Es
        # 这样的的形式返回，本别对应 requset.user 和 requset.auth

之后要在需要认证的某个视图类中加入authentication_classes以及在需要验证的对应的请求方法中添加对Token验证完成的判断逻辑

用户传递Token可以有多种方式：比如通过url参数或者请求头

通过url参数方式

user/views.py
class UserAddressListView(GenericAPIView, mixins.ListModelMixin):
    serializer_class = UserAddressSerializer
    queryset = UserAddress.objects
    authentication_classes = [JWTURLParamAuthentication,]

    def get(self, request):
        # print(request.user) # 对应的是JWTAuthentication的第一个返回值token_data
        # print(request.auth) # 对应的是JWTAuthentication的第二个返回值token
        # 将验证Token状态的逻辑放在视图类的好处是：当你在类中加入了authentication_classes后，如果有不需要验证的类，就不加验证的代码，就可以跳过验证进行相应
        if not request.user['status']:
            return ResponseMessage.UserResponse.failed(request.user)
        else:
            return ResponseMessage.UserResponse.success(request.user)

这个时候用过url参数的传递方式穿入Token，就会根据Token的过期时间和是否被篡改过进行响应了：

通过头信息传递参数

user/views.py
class UserAddressView(GenericAPIView,
                      mixins.CreateModelMixin,
                      mixins.RetrieveModelMixin,
                      mixins.UpdateModelMixin,
                      mixins.DestroyModelMixin):
    serializer_class = UserAddressSerializer
    queryset = UserAddress.objects
    authentication_classes = [JWTHeaderAuthentication,]

    def get_queryset(self):
        # 确保返回标准的Django QuerySet
        return UserAddress.objects.all()
    # self.get_queryset

    def post(self, request):
        return self.create(request)

    def get(self, request, pk):
        if not request.user['status']:
            return ResponseMessage.UserResponse.failed(request.user)
        else:
            return self.retrieve(request, pk)
            # return ResponseMessage.UserResponse.success(request.user)

    def put(self, request, pk):
        if not request.user['status']:
            return ResponseMessage.UserResponse.failed(request.user)
        else:
            return self.update(request, pk)

    def delete(self, request, pk):
        if not request.user['status']:
            return ResponseMessage.UserResponse.failed(request.user)
        else:
            return self.destroy(request,pk)

全局配置Toekn验证

在settings中可以直接全局配置Token验证，就无需再在一个一个views视图类中配置了

REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': ['utils.jwt.JWTURLParamAuthentication']
}

若有错误与不足请指出，关注DPT一起进步吧！！！