無情的學習機器

向之所欣，俯仰之間，已為陳跡...

DRF自定義系列

Posted on 2022-11-27 21:35 呱呱呱呱嘰里呱啦閱讀(98) 評論(0) 收藏舉報

django-rest-framework

django-rest-framework

頻率限制

# 新建頻率類繼承SimpleRateThrottle以重寫其內部的get_cache_key方法，并標明配置名稱
from rest_framework.throttling import SimpleRateThrottle

class MyThrottle(SimpleRateThrottle):
    scope = 'xxx'
    def get_cache_key(self, request, view):# return結果是頻率限制的key
        return request.META.get('REMOTE_ADDR')

# 局部配置或者全局配置頻率類
class BookView(ListAPIView):
    throttle_classes = [MyThrottle]
    queryset = models.Book.objects.all()
    serializer_class = BookModelSerializer
    pagination_class = MYLimitOffsetPagination
# 全局配置對應名稱的頻率值限制
REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_RATES': {
        'xxx': '3/m',
    },
}

自定義頻率

from rest_framework.throttling import BaseThrottle
# 自定義頻率類，需要重寫兩個方法，即allow_request、wait
class IPThrottle(BaseThrottle):
    visit_dict = {}
    def __init__(self):
        self.history_list = []
    def allow_request(self, request, view):
        ip = request.META.get('REMOTE_ADDR')
        ctime = time.time()
        if ip not in self.visit_dict:
            self.visit_dict[ip] = [ctime,]
            return True
        self.history_list = self.visit_dict[ip]
        while True:
            if ctime-self.history_list[-1]>60:
                self.history_list.pop()
            else:
                break
        if len(self.history_list)<2: # 配置一個周期內的頻次限制
            self.history_list.insert(0,ctime)
            return True
        else:
            return False
    def wait(self):
        ctime = time.time()
        return 60 - (ctime-self.history_list[-1])

from utils.throttle1 import IPThrottle
class BookView(ListAPIView):
    throttle_classes = [IPThrottle]
    queryset = models.Book.objects.all()
    serializer_class = BookModelSerializer
    pagination_class = MYLimitOffsetPagination

JWT

Json Web Token

header+payload+signature

djangorestframework-jwt模塊的使用

在繼承AbstractUser的用戶表模型下使用

# 安裝模塊
# 新建模型類繼承AbstractUser
# settings中配置認證用戶模型
AUTH_USER_MODEL = 'api.user'
# 創建超級用戶
# 配置路由，在登錄入口簽發token
from rest_framework_jwt.views import ObtainJSONWebToken
urlpatterns = [
    path('admin/', admin.site.urls),
    path('login/', ObtainJSONWebToken.as_view()),
]

# 以上是簡單使用，如果要在自定義視圖類中使用，可以在類中配置：
from rest_framework.permissions import IsAuthenticated

authentication_classes = [JSONWebTokenAuthentication,]
permission_classes = [IsAuthenticated,]

# 如果請求頭中沒有jwt標準格式的數據或者token不正確，則request.user為空，IsAuthenticated返回False，所以這里是組合使用才能生效

自定義登錄接口返回數據（由于之前是使用ObtainJSONWebToken類）

1.自寫登錄接口

2.用內置

# 源碼
def jwt_response_payload_handler(token, user=None, request=None):
    """
    Returns the response data for both the login and refresh views.
    Override to return a custom response such as including the
    serialized representation of the User.

    Example:

    def jwt_response_payload_handler(token, user=None, request=None):
        return {
            'token': token,
            'user': UserSerializer(user, context={'request': request}).data
        }

    """
    return {
        'token': token
    }
# 重寫jwt_response_payload_handler
def jwt_response_payload_handler(token, user=None, request=None):
    return {
        'token': token,
        'msg': '成功',
        'status': 100
    }
JWT_AUTH = {
    'JWT_RESPONSE_PAYLOAD_HANDLER':'api.auth.jwt_response_payload_handler'
} # settings配置

自定義認證（繼承BaseJSONWebTokenAuthentication）

# 新建認證類
from rest_framework import exceptions
from rest_framework_jwt.authentication import BaseJSONWebTokenAuthentication, jwt_decode_handler


class MyToken(BaseJSONWebTokenAuthentication):
    def authenticate(self, request):
        token = str(request.META.get('HTTP_AUTHORIZATION')) # 取出token
        try:
            payload = jwt_decode_handler(token) # 校驗是否過期，是否合法，是就返回荷載

        except Exception:
            raise exceptions.AuthenticationFailed('認證失敗')
        user = self.authenticate_credentials(payload)
        return user, token

# 局部配置
class BookView(APIView):
    authentication_classes = [MyToken,]
    def get(self, request):
        return Response('123')
# 測試，在GET請求頭配置key:AUTHORIZATION

# 自定義認證類的返回元組將作為request對象的兩個屬性

base64編解碼

base64.b64encode(bytes)

base64.b64decode(bytes)

開放media

from django.views.static import serve
from django.conf import settings
re_path('media/(?P<path>.*)',serve,{'document_root':settings.MEDIA_ROOT}) # serve即內置視圖函數

手動簽發token，完成多方式登錄

# 自定義序列化器
class LoginModelSerializer(serializers.ModelSerializer):
    username = serializers.CharField(max_length=16) # 覆蓋數據庫字段，避免字段校驗
    class Meta:
        model = models.User
        fields = ['username','password']
    def validate(self,attrs):
        username = attrs.get('username')
        password = attrs.get('password')
        if re.match('^1[3-9][0-9]{9}$',username):
            user=models.User.objects.filter(mobile=username).first()
        elif re.match('^.+@.+',username):
            user=models.User.objects.filter(email=username).first()
        else:
            user=models.User.objects.filter(name=username).first()
        if user:
            if user.check_password(password):
                payload = jwt_payload_handler(user)
                token = jwt_encode_handler(payload)
                self.context['token'] = token
                return attrs
            else:
                raise ValidationError('密碼錯誤')
        else:
            raise ValidationError('用戶不存在')


# 新建登錄視圖類
class LoginView(ViewSet):
    def login(self,request,*args,**kwargs):
        login_ser = ser.LoginModelSerializer(data=request.data,context={})
        login_ser.is_valid(raise_exception=True)
        token = login_ser.context.get('token')
        return Response({'status':100,'msg':'登錄成功','token':token})

序列化多條（many=True）時，自定義ListSerializer

# 自定義類繼承ListSerializer
class MyListSerializer(ListSerializer):
    def update(self, instance, validated_data):
        l = []
        for i in range(len(instance)):
            l.append(self.child.update(instance=instance[i],validated_data=validated_data[i]))
        return l

# 在序列化器類中配置list_serializer_class
class BookModelSerializer(serializers.ModelSerializer):
    # publish = serializers.CharField(source='publish.name') # 反序列化會有問題

    class Meta:
        list_serializer_class = MyListSerializer
        model = models.Book
        fields = ('name', 'price', 'authors', 'authors_list', 'publish', 'publish_name', )
        # depth = 0
        extra_kwargs = {
            'publish': {'write_only': True, },
            'publish_name': {'read_only': True, },
            'authors': {'write_only': True, },
            'authors_list': {'read_only': True, },
        }

# 視圖類中調用
instance_list = [models.Book.objects.filter(id=i.pop('id')).first() for i in request.data]
book_ser = BookModelSerializer(instance=instance_list,data=request.data,many=True)
book_ser.is_valid(raise_exception=True)
book_ser.save()
return Response(data=book_ser.data)

視圖函數給序列化類傳數據

book_ser = BookModelSerializer(instance=instance_list,data=request.data,many=True,context={'request':request})

認證組件：校驗用戶是否登錄

寫類繼承BaseAuthentication，重寫authenticate方法，內部寫認證邏輯，認證通過，返回兩個值一個值給request.user，另一個值給request.auth，認證失敗，拋出AuthenticationFailed異常

class MyAuthentication(BaseAuthentication):
    def authenticate(self, request):

        token = request.META.get('HTTP_TOKEN')
        if token:
            user = UserToken.objects.filter(token=token).first().user
            if user:
                return user,'ok'
            else:
                raise AuthenticationFailed('登錄信息校驗失敗')
        else:
            raise AuthenticationFailed('未登錄')
# 全局使用
from app01.Myauth import MyAuthentication
REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'MyAuthentication',

    ],
}
# 在具體視圖類中定義authentica_classes為空列表可以局部禁用

# 局部使用
# 在視圖類中定義authentication_classes=[]

權限：校驗用戶是否有權限進行后續操作

寫類繼承BasePermission，重寫has_permission方法，返回布爾值

# 定義權限類
from rest_framework.permissions import BasePermission
class UserAdminPermission(BasePermission):
    def has_permission(self, request, view):
        user = request.user
        # print(user.get_permissions_display())
        if user.permissions == 1: # 超級用戶
            return True
        else:
            return False

# 局部使用
class NewBooksView(ListAPIView, CreateAPIView):
    permission_classes = [UserAdminPermission]
    queryset = Book.objects.all()
    serializer_class = NewBookModelSerializer

# 全局使用
REST_FRAMEWORK = {
    'DEFAULT_PERMISSION_CLASSES': [
        'app01.123.UserAdminPermission',

    ],
}

# 局部禁用
permission_classes = []

參考

頻率：限制用戶訪問頻率

寫類繼承SimpleRateThrottle，具體見上詳述

解析器：即前段傳的編碼格式能不能解析，默認全配，全部可以解析

響應器：響應格式指定或者自動

過濾器：使用三方模塊django-filter

刷新頁面返回頂部