一、登录认证示例
模拟用户登录，获取token，当用户访问订单或用户中心时，判断用户携带正确的token，则允许查看订单和用户信息，否则抛出异常：
from django.conf.urls import url
from django.contrib import admin
from api import views urlpatterns = [
url(r'^admin/', admin.site.urls),
url(r'^api/v1/auth/$', views.AuthView.as_view()),
url(r'^api/v1/order/$', views.OrderView.as_view()),
url(r'^api/v1/userInfo/$', views.UserInfoView.as_view()),
]
urls.py
from django.db import models class UserInfo(models.Model):
user_type_choices = (
(1, "普通用户"),
(2, "vip"),
(3, "svip"),
)
user_type = models.IntegerField(choices=user_type_choices)
username = models.CharField(max_length=32, unique=True)
password = models.CharField(max_length=64) class UserToken(models.Model):
user = models.OneToOneField(to="UserInfo")
token = models.CharField(max_length=64) class Order(models.Model):
id = models.AutoField(primary_key=True)
name = models.CharField(max_length=32)
price = models.DecimalField(max_digits=5, decimal_places=2)
create_time = models.DateTimeField(auto_now_add=True)
user = models.ForeignKey(to="UserInfo", on_delete=models.CASCADE, null=True, blank=True)
models.py
import hashlib
import time from django.http import JsonResponse
from rest_framework.views import APIView
from rest_framework import exceptions from api import models def md5(user):
"""生成token"""
ctime = str(time.time()) # 当前时间
m = hashlib.md5(bytes(user, encoding="utf-8"))
m.update(bytes(ctime, encoding="utf-8"))
return m.hexdigest() class Authtication(object):
"""认证"""
def authenticate(self, request):
token = request._request.GET.get("token")
token_obj = models.UserToken.objects.filter(token=token).first()
if not token_obj:
raise exceptions.AuthenticationFailed("用户认证失败")
return (token_obj.user, token_obj) # rest framework会将这两个字段赋值给request，以供后续操作使用 def authenticate_header(self, request):
pass class AuthView(APIView):
"""登录"""
def post(self, request, *args, **kwargs):
res = {"code": 1000, "msg": None}
try:
# 从请求中获取用户登录信息
user = request._request.POST.get("username")
pwd = request._request.POST.get("password")
# 到数据库获取用户信息
user_obj = models.UserInfo.objects.filter(username=user, password=pwd).first()
# 如果获取到对象则说明认证成功，为登录用户创建token，如果认证失败则返回错误信息
if not user_obj:
res["code"] = 1001
res["msg"] = "用户名或密码错误"
else:
token = md5(user)
# 将token存入数据库：如果数据库存在token就更新，不存在就创建
models.UserToken.objects.update_or_create(user=user_obj, defaults={"token": token})
res["token"] = token
except Exception as e:
res["code"] = 1002
res["msg"] = e
return JsonResponse(res) class OrderView(APIView):
"""订单"""
authentication_classes = [Authtication] def get(self, request, *args, **kwargs):
res = {"code": 1000, "msg": None, "data": None}
try:
orders = models.Order.objects.filter(user=request.user).values("id", "name", "price", "create_time", "user__username")
res["data"] = list(orders)
except Exception as e:
res["code"] = 1002
res["msg"] = e
return JsonResponse(res) class UserInfoView(APIView):
"""用户中心"""
authentication_classes = [Authtication] def get(self, request, *args, **kwargs):
res = {"code": 1000, "msg": None, "data": None}
try:
# print(request.user) # 用户对象
# print(request.auth) # 认证对象
user = models.UserInfo.objects.filter(id=request.auth.user_id).values("id", "username", "password")
res["data"] = list(user)
except Exception as e:
res["code"] = 1002
res["msg"] = e
return JsonResponse(res)
views.py
二、rest framework认证流程源码
rest framework的request.py：
_not_authenticated()方法的处理流程：
三、rest framework配置
1、如何将之前写在视图里面的 authentication_classes 写入rest framework的配置文件中：
rest framework的配置信息在rest framework的settings.py里面：
rest framework的views.py：
新建一个utils文件，新建auth.py文件，将自定义的认证类写到这个文件里面：
代码：
from rest_framework import exceptions from api import models class Authtication(object):
"""认证"""
def authenticate(self, request):
token = request._request.GET.get("token")
token_obj = models.UserToken.objects.filter(token=token).first()
if not token_obj:
raise exceptions.AuthenticationFailed("用户认证失败")
return (token_obj.user, token_obj) # rest framework会将这两个字段赋值给request，以供后续操作使用 def authenticate_header(self, request):
pass
auth.py
再去项目的settings里面设置这个认证类的路径：
REST_FRAMEWORK = {
"DEFAULT_AUTHENTICATION_CLASSES": ["api.utils.auth.Authtication",]
}
settings.py
这样相当于做了一个全局配置，就不用在每个视图里面再去设置认证类了：
import hashlib
import time from django.http import JsonResponse
from rest_framework.views import APIView
# from rest_framework import exceptions from api import models def md5(user):
"""生成token"""
ctime = str(time.time()) # 当前时间
m = hashlib.md5(bytes(user, encoding="utf-8"))
m.update(bytes(ctime, encoding="utf-8"))
return m.hexdigest() class AuthView(APIView):
"""登录"""
authentication_classes = []
def post(self, request, *args, **kwargs):
res = {"code": 1000, "msg": None}
try:
# 从请求中获取用户登录信息
user = request._request.POST.get("username")
pwd = request._request.POST.get("password")
# 到数据库获取用户信息
user_obj = models.UserInfo.objects.filter(username=user, password=pwd).first()
# 如果获取到对象则说明认证成功，为登录用户创建token，如果认证失败则返回错误信息
if not user_obj:
res["code"] = 1001
res["msg"] = "用户名或密码错误"
else:
token = md5(user)
# 将token存入数据库：如果数据库存在token就更新，不存在就创建
models.UserToken.objects.update_or_create(user=user_obj, defaults={"token": token})
res["token"] = token
except Exception as e:
res["code"] = 1002
res["msg"] = e
return JsonResponse(res) class OrderView(APIView):
"""订单"""
# authentication_classes = [Authtication] def get(self, request, *args, **kwargs):
res = {"code": 1000, "msg": None, "data": None}
try:
orders = models.Order.objects.filter(user=request.user).values("id", "name", "price", "create_time", "user__username")
res["data"] = list(orders)
except Exception as e:
res["code"] = 1002
res["msg"] = e
return JsonResponse(res) class UserInfoView(APIView):
"""用户中心"""
# authentication_classes = [Authtication] def get(self, request, *args, **kwargs):
res = {"code": 1000, "msg": None, "data": None}
try:
# print(request.user) # 用户对象
# print(request.auth) # 认证对象
user = models.UserInfo.objects.filter(id=request.auth.user_id).values("id", "username", "password")
res["data"] = list(user)
except Exception as e:
res["code"] = 1002
res["msg"] = e
return JsonResponse(res)
views.py
2、匿名用户配置
看源码我们知道，当认证方法返回None时，rest framework默认是从它的配置文件中读取匿名用户来赋值给self.user：
当读取到“UNAUTHENTICATED_USER”这个值时，就会使用这个值，所有我们可以在项目的配置文件中对这个值进行设置：
这样，当认证方法返回None时，self.user="匿名用户"
那个UNAUTHENTICATED_TOKEN也是同样的设置方法。
四、rest framework内置的认证类
在rest framework的authentication.py中：
from rest_framework.authentication import BaseAuthentication class Authtication(BaseAuthentication):
"""自定制认证"""
def authenticate(self, request):
...... def authenticate_header(self, request):
......
BasicAuthentication认证类：是采用浏览器对用户名和密码进行base64加密，然后通过请求头发送给服务端，服务端获取请求头，对之前加密的用户名和密码进行解密，再到数据库进行校验。
五、rest framework权限使用
需求：给不同的视图设置不同的访问权限，如设置svip用户可以查看所有订单，普通用户和vip用户可以查看所有用户信息
1、源码实现流程
如果has_permission()返回True，则表示有权访问,否则无权访问。
2、权限控制的实现（局部）
创建权限类，在视图中使用
新建permission.py文件，写相关的权限控制类：
class MyPermissionSvip(object):
"""svip 访问权限控制"""
message = "只有svip用户才能访问" # 设置无权访问消息内容
def has_permission(self, request, view):
if request.user.user_type != 3: # 如果用户类型不是svip 则拒绝访问
return False
return True class MyPermissionOrdinaryAndVip(object):
"""普通用户和vip 访问权限控制"""
def has_permission(self, request, view):
if request.user.user_type == 3: # 如果用户类型是svip 则拒绝访问
return False
return True
permission.py
import hashlib
import time from django.http import JsonResponse
from rest_framework.views import APIView from api import models
from api.utils.permission import MyPermissionSvip, MyPermissionOrdinaryAndVip def md5(user):
"""生成token"""
ctime = str(time.time()) # 当前时间
m = hashlib.md5(bytes(user, encoding="utf-8"))
m.update(bytes(ctime, encoding="utf-8"))
return m.hexdigest() class AuthView(APIView):
"""登录"""
authentication_classes = [] def post(self, request, *args, **kwargs):
res = {"code": 1000, "msg": None}
try:
# 从请求中获取用户登录信息
user = request._request.POST.get("username")
pwd = request._request.POST.get("password")
# 到数据库获取用户信息
user_obj = models.UserInfo.objects.filter(username=user, password=pwd).first()
# 如果获取到对象则说明认证成功，为登录用户创建token，如果认证失败则返回错误信息
if not user_obj:
res["code"] = 1001
res["msg"] = "用户名或密码错误"
else:
token = md5(user)
# 将token存入数据库：如果数据库存在token就更新，不存在就创建
models.UserToken.objects.update_or_create(user=user_obj, defaults={"token": token})
res["token"] = token
except Exception as e:
res["code"] = 1002
res["msg"] = e
return JsonResponse(res) class OrderView(APIView):
"""订单"""
# authentication_classes = [Authtication] # 认证类（局部）
permission_classes = [MyPermissionSvip,] # 权限控制类（局部） def get(self, request, *args, **kwargs):
res = {"code": 1000, "msg": None, "data": None}
try:
orders = models.Order.objects.all().values("id", "name", "price", "create_time", "user__username")
res["data"] = list(orders)
except Exception as e:
res["code"] = 1002
res["msg"] = e
return JsonResponse(res) class UserInfoView(APIView):
"""用户中心"""
# authentication_classes = [Authtication]
permission_classes = [MyPermissionOrdinaryAndVip] def get(self, request, *args, **kwargs):
res = {"code": 1000, "msg": None, "data": None}
try:
# print(request.user) # 用户对象
# print(request.auth) # 认证对象
user = models.UserInfo.objects.all().values("id", "username", "password")
res["data"] = list(user)
except Exception as e:
res["code"] = 1002
res["msg"] = e
return JsonResponse(res)
views.py
3、权限控制的实现（全局）
在settings中导入权限类的路径来实现全局控制,这样就不需要在每个视图中设置permission_classes了。
import hashlib
import time from django.http import JsonResponse
from rest_framework.views import APIView from api import models
from api.utils.permission import MyPermissionSvip, MyPermissionOrdinaryAndVip def md5(user):
"""生成token"""
ctime = str(time.time()) # 当前时间
m = hashlib.md5(bytes(user, encoding="utf-8"))
m.update(bytes(ctime, encoding="utf-8"))
return m.hexdigest() class AuthView(APIView):
"""登录"""
authentication_classes = []
permission_classes = [] def post(self, request, *args, **kwargs):
res = {"code": 1000, "msg": None}
try:
# 从请求中获取用户登录信息
user = request._request.POST.get("username")
pwd = request._request.POST.get("password")
# 到数据库获取用户信息
user_obj = models.UserInfo.objects.filter(username=user, password=pwd).first()
# 如果获取到对象则说明认证成功，为登录用户创建token，如果认证失败则返回错误信息
if not user_obj:
res["code"] = 1001
res["msg"] = "用户名或密码错误"
else:
token = md5(user)
# 将token存入数据库：如果数据库存在token就更新，不存在就创建
models.UserToken.objects.update_or_create(user=user_obj, defaults={"token": token})
res["token"] = token
except Exception as e:
res["code"] = 1002
res["msg"] = e
return JsonResponse(res) class OrderView(APIView):
"""订单"""
# authentication_classes = [Authtication] # 认证类（局部）
# permission_classes = [MyPermissionSvip,] # 权限控制类（局部） def get(self, request, *args, **kwargs):
res = {"code": 1000, "msg": None, "data": None}
try:
orders = models.Order.objects.all().values("id", "name", "price", "create_time", "user__username")
res["data"] = list(orders)
except Exception as e:
res["code"] = 1002
res["msg"] = e
return JsonResponse(res) class UserInfoView(APIView):
"""用户中心"""
# authentication_classes = [Authtication]
permission_classes = [MyPermissionOrdinaryAndVip] def get(self, request, *args, **kwargs):
res = {"code": 1000, "msg": None, "data": None}
try:
# print(request.user) # 用户对象
# print(request.auth) # 认证对象
user = models.UserInfo.objects.all().values("id", "username", "password")
res["data"] = list(user)
except Exception as e:
res["code"] = 1002
res["msg"] = e
return JsonResponse(res)
views.py
六、rest framework内置的权限类
from rest_framework.permissions import BasePermission
在rest framework的permissions.py源码中：
建议在自定义权限控制类时，继承这个BasePermission类：
七、rest framework的访问频率控制
如：限制某个用户1分钟只能访问多少次
1、源码流程
如果allow_request()返回True，表示可以访问，否则表示频率太高，不能访问
2、需求：对用户登录进行频率控制
新建文件：
代码：
import time
from rest_framework.throttling import BaseThrottle VISIT_RECORD = {} # 存储用户访问记录 class VisitThrotlle(BaseThrottle):
"""用户登录访问频率控制"""
def __init__(self):
self.history = None def allow_request(self, request, view):
# 获取用户IP
# remote_addr = request.META.get('REMOTE_ADDR')
remote_addr = self.get_ident(request) # 也可以通过继承父类方法来获取IP ctime = time.time() # 用户访问时间
# 判断用户是否可以访问 如果该IP还没有访问过，直接放行；
# 如果该IP已经存在于访问记录中，判断其访问频率是否达到上限
if remote_addr not in VISIT_RECORD:
VISIT_RECORD[remote_addr] = [ctime]
return True
self.history = VISIT_RECORD.get(remote_addr) # 获取访问历史时间列表
# 当前访问时间与访问记录中的时间进行比较，如果当前时间是在一分钟之后访问的，就删掉访问记录中的时间
while self.history and self.history[-1] < ctime-60:
self.history.pop()
# 控制一分钟内允许访问3次
if len(self.history) < 3:
self.history.insert(0, ctime) # 将最近的一次访问时间插入到列表第一个位置
return True
return False def wait(self):
# 可以返回None，也可以返回时间，提示用户还要等多少秒就可以访问了
ctime = time.time()
return 60 - (ctime - self.history[-1])
throtlle.py
在views.py中引入：
import hashlib
import time from django.http import JsonResponse
from rest_framework.views import APIView from api import models
from api.utils.permission import MyPermissionSvip, MyPermissionOrdinaryAndVip
from api.utils.throtlle import VisitThrotlle def md5(user):
"""生成token"""
ctime = str(time.time()) # 当前时间
m = hashlib.md5(bytes(user, encoding="utf-8"))
m.update(bytes(ctime, encoding="utf-8"))
return m.hexdigest() class AuthView(APIView):
"""登录"""
authentication_classes = []
permission_classes = []
throttle_classes = [VisitThrotlle] # 访问频率控制 def post(self, request, *args, **kwargs):
res = {"code": 1000, "msg": None}
try:
# 从请求中获取用户登录信息
user = request._request.POST.get("username")
pwd = request._request.POST.get("password")
# 到数据库获取用户信息
user_obj = models.UserInfo.objects.filter(username=user, password=pwd).first()
# 如果获取到对象则说明认证成功，为登录用户创建token，如果认证失败则返回错误信息
if not user_obj:
res["code"] = 1001
res["msg"] = "用户名或密码错误"
else:
token = md5(user)
# 将token存入数据库：如果数据库存在token就更新，不存在就创建
models.UserToken.objects.update_or_create(user=user_obj, defaults={"token": token})
res["token"] = token
except Exception as e:
res["code"] = 1002
res["msg"] = e
return JsonResponse(res) class OrderView(APIView):
"""订单"""
# authentication_classes = [Authtication] # 认证类（局部）
# permission_classes = [MyPermissionSvip,] # 权限控制类（局部） def get(self, request, *args, **kwargs):
res = {"code": 1000, "msg": None, "data": None}
try:
orders = models.Order.objects.all().values("id", "name", "price", "create_time", "user__username")
res["data"] = list(orders)
except Exception as e:
res["code"] = 1002
res["msg"] = e
return JsonResponse(res) class UserInfoView(APIView):
"""用户中心"""
# authentication_classes = [Authtication]
permission_classes = [MyPermissionOrdinaryAndVip] def get(self, request, *args, **kwargs):
res = {"code": 1000, "msg": None, "data": None}
try:
# print(request.user) # 用户对象
# print(request.auth) # 认证对象
user = models.UserInfo.objects.all().values("id", "username", "password")
res["data"] = list(user)
except Exception as e:
res["code"] = 1002
res["msg"] = e
return JsonResponse(res)
views.py
3、频率控制也可以做全局设置，方法与权限控制相同
4、内置控制频率的类
在rest_framework的throttling.py中:
1、示例1
在throtlle.py中继承SimpleRateThrottle类：
from rest_framework.throttling import BaseThrottle, SimpleRateThrottle class VisitThrotlle(SimpleRateThrottle):
"""用户登录访问频率控制"""
scope = "throtlle_rate" # 定义一个key，从配置文件中获取访问频次 def get_cache_key(self, request, view):
# 程序回去Django的缓存中获取key，此时我们重写这个方法，给他返回一个用户IP作为key
return self.get_ident(request)
throtlle.py
这个实现效果和上面自己写的一样。
2、示例2
对登录用户做频率控制
from rest_framework.throttling import BaseThrottle, SimpleRateThrottle class UserThrotlle(SimpleRateThrottle):
"""对已登录用户进行访问频率控制"""
scope = "user_throtlle_rate" # 定义一个key，从配置文件中获取访问频次 def get_cache_key(self, request, view):
# 返回一个用户唯一标志，如用户名
return request.user.username class VisitThrotlle(SimpleRateThrottle):
"""匿名用户登录访问频率控制"""
scope = "throtlle_rate" # 定义一个key，从配置文件中获取访问频次 def get_cache_key(self, request, view):
# 程序回去Django的缓存中获取key，此时我们重写这个方法，给他返回一个用户IP作为key
return self.get_ident(request)
throtlle.py
setting.py：
