1.登录成功后返回token,这里使用authenticate进行校验是否存在该用户
def login(request):try:data = json.loads(request.body)username = data.get('username')password = data.get('password')if not all([username, password]):return to_response(status=400, msg='参数不完整!')user = authenticate(username=username, password=password)if user:expiration_date = datetime.utcnow() + timedelta(seconds=settings.TOKEN_EXPIRED_TIME)payload = {'id': user.id, 'username': user.username, 'exp': expiration_date}token = encode(payload, settings.SECRET_KEY, algorithm='HS256')data = {'user': {'id': user.id, 'username': user.username},'token': token,}return to_response(status=200, data=data, msg='登录成功!')else:return to_response(status=400, msg='用户名或密码错误!')except Exception as e:return to_response(status=500, msg=str(e))
2.定义自定义认证后端,并在其中校验token
from django.contrib.auth.backends import BaseBackend
from users.models import UserProfile
from django.conf import settings
from jwt import decodeclass MyModelBackend(BaseBackend):def authenticate(self, request, token=None):if token:token_info = decode(token, settings.SECRET_KEY, algorithms=['HS256'])user_id = token_info.get('id')user = UserProfile.objects.get(id=user_id)if user:return userelse:return Noneelse:return Nonedef get_user(self, user_id):try:return UserProfile.objects.get(pk=user_id)except UserProfile.DoesNotExist:return None##################在settings中配置######################################
AUTHENTICATION_BACKENDS = ['django.contrib.auth.backends.ModelBackend', # 默认的 ModelBackend 也可以保留'django325.utils.MyModelBackend.MyModelBackend', # 自定义的 ModelBackend'
]
3.分装装饰器
from django.contrib.auth import authenticate
from jwt import InvalidTokenError, ExpiredSignatureErrorfrom django325.utils.response import to_response
from functools import wrapsdef token_required(func):@wraps(func)def wrapper(request, *args, **kwargs):token = request.headers.get('Authorization')if token:try:user = authenticate(request=request, token=token)except ExpiredSignatureError:return to_response(status=401, msg='token已过期!')except InvalidTokenError:return to_response(status=401, msg='token验证失败')except Exception as e:return to_response(status=401, msg=str(e))if user:request.user = userreturn func(request, *args, **kwargs)# 用户未认证,返回错误响应return to_response(status=401, msg='用户未认证')return wrapper
4.使用
@token_required
def test(request):return to_response(status=200, msg='测试成功!')