补充
2 认证相关：
session cookie token 认证相关的 这里用token
token 1 有时间限制，超时则失效 2 每次登录更换一个token
3 访问频率限制
1 防止同一时间多次访问的黑客攻击，增加服务器压力。
4 我们的解析 认证 权限 频率 是在用户get post等请求之前就做好了的
5 uuid随机数模块，uuid.uuid4()获取随机数
认证 权限 频率组件操作流程
认证需要建立user表和token表
# 用户表
class User(models.Model):
user=models.CharField(max_length=32)
pwd=models.CharField(max_length=32)
type=((1,"VIP"),(2,"SVIP"),(3,"SSSVIP"))
user_type=models.IntegerField(choices=type) # token表
class UserToken(models.Model):
user=models.OneToOneField("User")
token=models.CharField(max_length=128)
用户登录视图添加token
from app01.models import User,UserToken class LoginView(APIView):
"""
1000:成功
1001:用户名或者密码错误
1002:异常错误
"""
def post(self,request): response = {"code": 1000, "msg": None, "user": None}
try:
print(request.data)
user = request.data.get("user")
pwd = request.data.get("pwd") user = User.objects.filter(user=user, pwd=pwd).first()
import uuid
random_str = uuid.uuid4()
if user: UserToken.objects.update_or_create(user=user, defaults={"token": random_str})
response["user"] = user.user
response["token"] = random_str
else:
response["code"] = 1001
response["msg"] = "用户名或者密码错误" except Exception as e:
response["code"]=1002
response["msg"]=str(e) return Response(response)
认证自定义的模块
其中继承 BaseAuthentication除了类本生加的功能外还继承了header方法，如果不继承需要自己写一个header方法，否则会报错！
from app01.models import UserToken
from rest_framework.exceptions import AuthenticationFailed from rest_framework.authentication import BaseAuthentication class UserAuth(BaseAuthentication): def authenticate(self,request): token=request.query_params.get("token") usertoken=UserToken.objects.filter(token=token).first()
if usertoken:
return usertoken.user,usertoken.token
else:
raise AuthenticationFailed("认证失败！")
权限组件自定义的模块
from rest_framework.permissions import AllowAny class SVIPPermission(object):
message="您没有访问权限！"
def has_permission(self,request,view):
if request.user.user_type >= 2:
return True
return False
频率组件自定义的模块
以一分钟访问3次为例
from rest_framework.throttling import BaseThrottle
VISIT_RECORD={}
class VisitThrottle(BaseThrottle):
def __init__(self):
self.history=None
def allow_request(self,request,view):
remote_addr = request.META.get('REMOTE_ADDR')
print(remote_addr)
import time
ctime=time.time()
if remote_addr not in VISIT_RECORD:
VISIT_RECORD[remote_addr]=[ctime,]
return True
history=VISIT_RECORD.get(remote_addr)
self.history=history
while history and history[-1]<ctime-60:
history.pop()
if len(history)<3:
history.insert(0,ctime)
return True
else:
return False
def wait(self):
import time
ctime=time.time()
return 60-(ctime-self.history[-1])
在全局中的定制
REST_FRAMEWORK={
# 解析器组件
'DEFAULT_PARSER_CLASSES': (
'rest_framework.parsers.JSONParser',
'rest_framework.parsers.FormParser',
),
# 认证组件
'DEFAULT_AUTHENTICATION_CLASSES': (
'app01.utils.auth_class.UserAuth',
),
# 权限组件
'DEFAULT_PERMISSION_CLASSES': (
'app01.utils.permission_class.SVIPPermission',
),
# 频率组件
'DEFAULT_THROTTLE_CLASSES': (),
}
在局部中的定制及在视图中的使用
# 导入自定义的认证功能类
from app01.utils.auth_class import UserAuth
# 导入自定义的权限功能类
from app01.utils.permission_class import SVIPPermission
# 导入自定义的频率功能类
from app01.utils.throttle_classes import VisitThrottle from rest_framework.throttling import BaseThrottle
class VisitThrottle(BaseThrottle):
def allow_request(self,request,view):
"""
限制IP每分钟访问不能超过3次
:param request:
:param view:
:return:
"""
print(self.get_ident(request))
remote_addr = request.META.get('REMOTE_ADDR')
print("REMOTE_ADDR",remote_addr)
return False class BookView(APIView):
# 认证
# authentication_classes = [UserAuth]
# 权限
# permission_classes = [SVIPPermission]
# 频率
throttle_classes = [VisitThrottle] def get(self,request):
'''
查看所有书籍
:param request:
:return:
''' print(request.user,request.auth) book_list=Book.objects.all()
serializer=BookSerializer(book_list,many=True)
return Response(serializer.data) def post(self,request):
'''
添加一条书籍
:param request:
:return:
'''
print(request.data) serializer=BookSerializer(data=request.data,many=False) if serializer.is_valid():
serializer.save() # create操作 return Response(serializer.data)
else:
return Response(serializer.errors) class SBookView(APIView): def get(self,request,id):
edit_obj=Book.objects.get(pk=id)
serializer=BookSerializer(edit_obj,many=False)
return Response(serializer.data) def put(self,request,id):
edit_obj = Book.objects.get(pk=id)
serializer=BookSerializer(data=request.data,instance=edit_obj)
if serializer.is_valid():
serializer.save() # edit_obj.update(request.data)
return Response(serializer.data)
else:
return Response(serializer.errors) def delete(self,request,id):
edit_obj = Book.objects.get(pk=id).delete()
return Response("")
以认证为例的源码解析
//用户访问，当经过dispath时
1 def dispatch(self, request, *args, **kwargs):
self.initial(request, *args, **kwargs)
initial()是我们的解析 认证 权限 频率功能
2 在initial函数中
def initial(self, request, *args, **kwargs):
# 执行认证功能
self.perform_authentication(request)
# 执行权限功能
self.check_permissions(request)
# 执行频率功能
self.check_throttles(request)
3 我们走perform_authentication(request)认证功能
def perform_authentication(self, request):
request.user
从这里我们要从request实例中找user方法
4 我们通过request实例对象找到它的类中的user方法
def dispatch(self, request, *args, **kwargs):
request = self.initialize_request(request, *args, **kwargs)
def initialize_request(self, request, *args, **kwargs)
return Request(
request,
parsers=self.get_parsers(),
authenticators=self.get_authenticators(),
negotiator=self.get_content_negotiator(),
parser_context=parser_context
)
class Request(object):
def user(self):
if not hasattr(self, '_user'):
with wrap_attributeerrors():
self._authenticate()
return self._user
5 在_authenticate()中：
def _authenticate(self):
# 和解析器一样，获得[UserAuth()] 获得顺序：当前视图类下-->全局setting-->默认default
for authenticator in self.authenticators:
try:
# 执行authenticate方法进行验证，返回元组/空 或抛出一个异常
user_auth_tuple = authenticator.authenticate(self)
except exceptions.APIException:
self._not_authenticated()
raise
# 如果返回一个元组则执行下面的语句
if user_auth_tuple is not None:
self._authenticator = authenticator
self.user, self.auth = user_auth_tuple
# 这里有个坑，如果返回元组，则直接结束_authenticate函数，不继续循环了！
return
这里如果不抛异常则进入下一个组件
