drf-jwt源码执行流程(了解)
签发(登录)源码分析
登录接口,路由匹配成功,执行obtain_jwt_token ----》post请求----》ObtainJSONWebToken的post方法
path('login/',obtain_jwt_token),
# ObtainJSONWebToken的post方法 继承APIView
def post(self, request, *args, **kwargs):
# 实例化得到序列化类
serializer = self.get_serializer(data=request.data)
# 做 校验:字段自己,局部钩子,全局钩子
if serializer.is_valid():
# user:当前登录用户
user = serializer.object.get('user') or request.user
# 签发的token
token = serializer.object.get('token')
# 构造返回格式,咱们可以自定制---- 》讲过
response_data = jwt_response_payload_handler(token, user, request)
response = Response(response_data)
if api_settings.JWT_AUTH_COOKIE:
expiration = (datetime.utcnow() +
api_settings.JWT_EXPIRATION_DELTA)
response.set_cookie(api_settings.JWT_AUTH_COOKIE,
token,
expires=expiration,
httponly=True)
# 最终返回了定制的返回格式
return response
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
如何得到user,如何签发的token----》在序列化类的全局钩子中得到的user和签发的token
JSONWebTokenSerializer---全局钩子---validate
前端传入,校验过后的数据----》{"username":"lqz","password":"lqz12345"}
def validate(self, attrs):
credentials = {
# self.username_field: attrs.get(self.username_field),
'username':attrs.get('username'),
'password': attrs.get('password')
}
if all(credentials.values()):
# auth模块,authenticate可以传入用户名,密码如果用户存在,就返回用户对象,如果不存在就是None
# 正确的用户
user = authenticate(**credentials)
if user:
# 校验用户是否是活跃用户,如果禁用了,不能登录成功
if not user.is_active:
msg = _('User account is disabled.')
raise serializers.ValidationError(msg)
# 荷载---》通过user得到荷载{id,name,email,exp}
payload = jwt_payload_handler(user)
return {
# jwt_encode_hangdler通过荷载得到token串
'token': jwt_encode_handler(payload),
'user': user
}
else:
msg = _('Unable to log in with provided credentials.')
raise serializers.ValidationError(msg)
else:
msg = _('Must include "{username_field}" and "password".')
msg = msg.format(username_field=self.username_field)
raise serializers.ValidationError(msg)
'''
重点
1.通过user得到荷载:payload = jwt_payload_handler(user)
2.通过荷载签发token:jwt_encode_handler(payload)
'''
了解:
翻译函数,只要做了国际化,放的英文,会翻译成该国语言(配置文件配置)
from django.utils.translation import ugettext as _
msg = _('Unable to log in with provided credentials.')
认证(认证类)源码分析
JSONWebTokenAuthentication----》父类 BaseJSONWebTokenAuthentication---》authenticate方法
def authenticate(self, request):
# 前端带在请求头中的token值
jwt_value = self.get_jwt_value(request)
# 如果没有携带token 就不校验了
if jwt_value is None:
return None
try:
# jwt_value就是token
# 通过token得到荷载,中途会出错
# 出错的原因:
篡改token
过期
未知错误
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()
# 如果能顺利解开,没有被异常捕获,说明token是可以信任的
# payload就可以使用,通过payload得到当前登录用户
user = self.authenticate_credentials(payload)
# 返回当前登录用户,token
return (user, jwt_value)
# jwt_value = self.get_jwt)value(request)
def get_jwt_value(self, request):
# 拿到了前端请求头中传入的jwt xxxxxxxxx
# auth = [jwt,xxxxxxx]
auth = get_authorization_header(request).split()
# 'jwt'
auth_header_prefix = api_settings.JWT_AUTH_HEADER_PREFIX.lower()
if not auth:
# 请求头中如果没带 去cookie中取
if api_settings.JWT_AUTH_COOKIE:
return request.COOKIES.get(api_settings.JWT_AUTH_COOKIE)
return None
if smart_text(auth[0].lower()) != auth_header_prefix:
return None
if len(auth) == 1:
msg = _('Invalid Authorization header. No credentials provided.')
raise exceptions.AuthenticationFailed(msg)
elif len(auth) > 2:
msg = _('Invalid Authorization header. Credentials string '
'should not contain spaces.')
raise exceptions.AuthenticationFailed(msg)
return auth[1]
# 认证类配置了,如果不传jwt 不会校验 必须配合权限类一起使用
自定义用户表签发和认证
签发代码
视图类
from rest_framework_jwt.settings import api_settings
jwt_payload_handler = api_settings.JWT_PAYLOAD_HANDLER
jwt_encode_handler = api_settings.JWT_ENCODE_HANDLER
from rest_framework.viewsets import ViewSet
from rest_framework.decorators import action
from rest_framework.response import Response
from .models import UserInfo
class UserView(ViewSet):
@action(methods=['POST'], detail=False)
def login(self, request, *args, **kwargs):
username = request.data.get('username')
password = request.data.get('password')
user = UserInfo.objects.filter(username=username, password=password).first()
if user:
# 登录成功,签发token
# 通过user得到payload
payload = jwt_payload_handler(user)
# 通过payload得到token
token = jwt_encode_handler(payload)
return Response({'code': 10000, 'msg': '登录成功', 'token': token})
else:
return Response({'code': 10001, 'msg': '用户名或密码错误'})
模型类
class UserInfo(models.Model):
username = models.CharField(max_length=32)
password = models.CharField(max_length=32)
路由
from django.contrib import admin
from django.urls import path
from rest_framework.routers import SimpleRouter
from app01.views import UserView
router = SimpleRouter()
router.register('user',UserView,'user')
urlpatterns = [
path('admin/', admin.site.urls),
]
urlpatterns += router.urls
认证代码
认证类
from rest_framework_jwt.settings import api_settings
jwt_decode_handler = api_settings.JWT_DECODE_HANDLER
from rest_framework.authentication import BaseAuthentication
from rest_framework.exceptions import AuthenticationFailed
import jwt
from .models import UserInfo
class JsonWebTokenAuthentication(BaseAuthentication):
def authenticate(self, request):
# 取出token ----》请求头中,就叫token
token = request.META.get('HTTP_TOKEN')
if token:
try:
payload = jwt_decode_handler(token)
# 得到当前登录用户
# user = UserInfo.objects.get(pk=payload.get('user_id'))
# 只要访问一次需要登录的接口 就会去UserInfo 表中查一次用户----》优化
user = UserInfo(id=payload.get('user_id'), username=payload.get('username'))
# user= {'id':payload.get('user_id')}
return user, token
except jwt.ExpiredSignature:
raise AuthenticationFailed('token过期')
except jwt.DecodeError:
raise AuthenticationFailed('token认证失败')
except jwt.InvalidTokenError:
raise AuthenticationFailed('token无效')
except Exception as e:
raise AuthenticationFailed('未知异常')
raise AuthenticationFailed('token没有传,认证失败')
视图类
from rest_framework_jwt.settings import api_settings
jwt_payload_handler = api_settings.JWT_PAYLOAD_HANDLER
jwt_encode_handler = api_settings.JWT_ENCODE_HANDLER
from .authentication import JsonWebTokenAuthentication
from rest_framework.viewsets import ViewSet
from rest_framework.decorators import action
from rest_framework.response import Response
from .models import UserInfo
class UserView(ViewSet):
@action(methods=['POST'], detail=False)
def login(self, request, *args, **kwargs):
username = request.data.get('username')
password = request.data.get('password')
user = UserInfo.objects.filter(username=username, password=password).first()
if user:
# 登录成功,签发token
# 通过user得到payload
payload = jwt_payload_handler(user)
# 通过payload得到token
token = jwt_encode_handler(payload)
return Response({'code': 10000, 'msg': '登录成功', 'token': token})
else:
return Response({'code': 10001, 'msg': '用户名或密码错误'})
class TestView(ViewSet):
authentication_classes = [JsonWebTokenAuthentication]
@action(methods=['GET'], detail=False)
def test(self, request):
return Response('ok')
路由
from django.contrib import admin
from django.urls import path
from rest_framework.routers import SimpleRouter
from app01.views import UserView, TestView
router = SimpleRouter()
router.register('user', UserView, 'user')
# 127.0.0.1:8000/test/test/
router.register('test', TestView, 'test')
urlpatterns = [
path('admin/', admin.site.urls),
]
urlpatterns += router.urls
simpleui的使用
公司里做项目,要使用权限,要快速搭建后台管理,使用django的admin直接搭建,django的admin界面不好
第三方的美化:
xadmin:作者不维护了 bootstrap+jquery
simpleui:vue,界面更好看
现阶段,一般前后端分离比较多:django+vue
带权限的前后端分离的快速开发框架
django-vue-admin
使用步骤
1.安装
pip3.8 installdjango-simpleui -i http://mirrors.aliyun.com/pypi/simple/ --trusted
-host mirrors.aliyun.com
2.在app中注册
3.调整左侧导航栏 ----》
menu_display对应menus name
如果是项目的app 就menus写app
菜单可以多级,一般内部app都是一级
可以增加app外的其他链接,跟之前前后端混合项目一样的写法: show的路由
SIMPLEUI_CONFIG = {
'system_keep': False,
'menu_display': ['图书管理', '权限认证', '张红测试'], # 开启排序和过滤功能, 不填此字段为默认排序和全部显示, 空列表[] 为全部不显示.
'dynamic': True, # 设置是否开启动态菜单, 默认为False. 如果开启, 则会在每次用户登陆时动态展示菜单内容
'menus': [
{
'name': '图书管理',
'app': 'app01',
'icon': 'fas fa-code',
'models': [
{
'name': '图书',
'icon': 'fa fa-user',
'url': 'app01/book/'
},
{
'name': '出版社',
'icon': 'fa fa-user',
'url': 'app01/publisssh/'
},
{
'name': '作者',
'icon': 'fa fa-user',
'url': 'app01/author/'
},
{
'name': '作者详情',
'icon': 'fa fa-user',
'url': 'app01/authordetail/'
},
]
},
{
'app': 'auth',
'name': '权限认证',
'icon': 'fas fa-user-shield',
'models': [
{
'name': '用户',
'icon': 'fa fa-user',
'url': 'auth/user/'
},
{
'name': '组',
'icon': 'fa fa-user',
'url': 'auth/group/'
},
]
},
{
'name': '张红测试',
'icon': 'fa fa-file',
'models': [
{
'name': 'Baidu',
'icon': 'far fa-surprise',
# 第三级菜单 ,
'models': [
{
'name': '爱奇艺',
'url': 'https://www.iqiyi.com/dianshiju/'
# 第四级就不支持了,element只支持了3级
}, {
'name': '百度问答',
'icon': 'far fa-surprise',
'url': 'https://zhidao.baidu.com/'
}
]
},
{
'name': '大屏展示',
'url': '/show/',
'icon': 'fab fa-github'
}]
}
]
}
4.内部app,图书管理系统 某个链接要展示的字段---》在admin.py中---》自定义按钮
@admin.register(Book)
class BookAdmin(admin.ModelAdmin):
list_display = ('nid', 'name', 'price', 'publish_date', 'publish')
# 增加自定义按钮
actions = ['custom_button']
def custom_button(self, request, queryset):
print(queryset)
custom_button.confirm = '你是否执意要点击这个按钮?'
# 显示的文本,与django admin一致
custom_button.short_description = '测试按钮'
# icon,参考element-ui icon与https://fontawesome.com
# custom_button.icon = 'fas fa-audio-description'
# # 指定element-ui的按钮类型,参考https://element.eleme.cn/#/zh-CN/component/button
custom_button.type = 'danger'
# # 给按钮追加自定义的颜色
# custom_button.style = 'color:black;'
5.app名字显示中文,字段名字显示中文
新增,查看修改展示中文,在表模型的字段上加,verbose_name='图书名字',help_text='这里填图书名'
app名字中文:apps.py---》verbose_name= '图书管理系统'
6.其他配置项
SIMPLEUI_LOGIN_PARTICLES = False #登录页面动态效果
SIMPLEUI_LOGO = 'https://avatars2.githubusercontent.com/u/13655483?s=60&v=4'#图标替换
SIMPLEUI_HOME_INFO = False #首页右侧github提示
SIMPLEUI_HOME_QUICK = False #快捷操作
SIMPLEUI_HOME_ACTION = False # 动作
标签:自定义,get,jwt,token,源码,user,import,payload,drf
From: https://www.cnblogs.com/xm15/p/17110286.html