django drf 统一处理操作人和时间字段
场景
如果你的表结构有这些字段：创建人/创建时间/更新人/更新时间/删除人/删除时间，我们可以统一处理这些字段的更新和插入，而不需要额外显示操作。
代码
1.ActionViewSetMixin：
import datetime
from rest_framework import status
from rest_framework.response import Response
class ActionViewSetMixin:
def destroy_mixin(self, request):
instance = self.get_object()
data = {
"deleted_by": request.user.username,
"deleted_at": datetime.datetime.now(),
"is_deleted": True,
}
serializer = self.get_serializer(instance, data=data, partial=True)
serializer.is_valid(raise_exception=True)
self.perform_update(serializer)
if getattr(instance, "_prefetched_objects_cache", None):
# If 'prefetch_related' has been applied to a queryset, we need to
# forcibly invalidate the prefetch cache on the instance.
instance._prefetched_objects_cache = {}
return Response(serializer.data)
def destroy(self, request, *args, **kwargs):
return self.destroy_mixin(request)
def update_mixin(self, request, serializer):
serializer.is_valid(raise_exception=True)
data = serializer.validated_data
data["updated_by"] = request.user.username
data["updated_at"] = datetime.datetime.now()
partial = True
instance = self.get_object()
serializer = self.get_serializer(instance, data=data, partial=partial)
serializer.is_valid(raise_exception=True)
self.perform_update(serializer)
if getattr(instance, "_prefetched_objects_cache", None):
# If 'prefetch_related' has been applied to a queryset, we need to
# forcibly invalidate the prefetch cache on the instance.
instance._prefetched_objects_cache = {}
return Response(serializer.data)
def update(self, request, *args, **kwargs):
serializer_class = kwargs.pop("serializer", self.get_serializer)
return self.update_mixin(request, serializer_class(data=request.data))
def create_mixin(self, request, *args, **kwargs):
data = request.data.copy()
data["created_by"] = request.user.username
data["created_at"] = datetime.datetime.now()
serializer_class = kwargs.pop("serializer", self.get_serializer)
serializer = serializer_class(data=data)
serializer.is_valid(raise_exception=True)
self.perform_create(serializer)
headers = self.get_success_headers(serializer.data)
return Response(
serializer.data, status=status.HTTP_201_CREATED, headers=headers
)
def create(self, request, *args, **kwargs):
return self.create_mixin(request, *args, **kwargs)2.APIModelViewSet继承ModelViewSet和ActionViewSetMixin：
from rest_framework.viewsets import ModelViewSet
from common.drf.mixins import ActionViewSetMixin, ResponseMixin
class APIModelViewSet(ActionViewSetMixin, ResponseMixin, ModelViewSet):
extra_permissions = [] # 不覆盖默认的权限
action_extra_permission_mapping = {}
def parse_page_params(self):
if self.request.method == "GET":
page = self.request.query_params.get("page", 1)
page_size = self.request.query_params.get("page_size", 10)
else:
page = self.request.data.get("page", 1)
page_size = self.request.data.get("page_size", 10)
try:
return int(page), int(page_size)
except Exception as e:
return 1, 10
def get_permissions(self):
permissions = super(APIModelViewSet, self).get_permissions()
extra = [x() for x in self.extra_permissions]
action = self.action_extra_permission_mapping.get(self.action, [])
return permissions + extra + [p() for p in action]
@classmethod
def get_validated_params(cls, serializer, params):
s = serializer(data=params)
s.is_valid(raise_exception=True)
return s.validated_data3.视图类继承APIModelViewSet：
class ScanRecordViewSet(APIModelViewSet):
queryset = models.ScanRecord.exclude_deleted_objects.all()
serializer_class = serializers.ScanRecordSerializer
pagination_class = DataPageNumberPagination
filter_backends = [CustomFilterBackend, SearchFilter]
search_fields = ["status", "mode", 'risk']
custom_query_fields = ["status", "mode", "risk"]