模型基类

class BaseEntity(models.Model):
    """抽象基类，用于提供创建人、更新时间等公共字段."""

    create_by = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        null=True,
        on_delete=models.SET_NULL,
        verbose_name="创建人",
        related_name="create+",
    )
    update_by = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        null=True,
        on_delete=models.SET_NULL,
        verbose_name="更新人",
        related_name="updated+",
    )
    update_date = models.DateTimeField(auto_now=True, verbose_name="更新时间")
    create_date = models.DateTimeField(auto_now_add=True, verbose_name="创建时间", editable=False, db_index=True)
    remark = models.CharField(max_length=500, verbose_name="备注", null=True, blank=True)
    del_flag = models.BooleanField(default=False, verbose_name="删除", editable=False)

    class Meta:
        ordering = ["-id"]  # 默认按 ID 降序排列
        abstract = True  # 声明为抽象类，不会在数据库中创建表

CRUD基类

class CoreViewSet(viewsets.ModelViewSet):
    """
    增删改查API基类，提供基本的CRUD操作和自定义响应
    """

    queryset = ""
    filter_fields = ()
    search_fields = ()
    serializer_class = ""
    permission_classes = ()
    filter_class = QueryFilter
    pagination_class = LargePagination
    # http_method_names = ["POST", "DELETE", "PUT", "GET"]  # 仅允许这些方法
    filter_backends = (DjangoFilterBackend, SearchFilter, OrderingFilter)

    def create(self, request, *args, **kwargs):
        """
        创建新的对象，并设置创建人和更新人。
        """
        # 设置创建人、更新人
        request.data["create_by"] = request.user.id
        request.data["update_by"] = request.user.id

        # 校验数据，保存对象
        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        self.perform_create(serializer)

        headers = self.get_success_headers(serializer.data)
        return JsonResult(
            data=serializer.data, msg="success", code=201, status=status.HTTP_201_CREATED, headers=headers
        )

    def destroy(self, request, *args, **kwargs):
        """
        删除指定的对象。
        """
        instance = self.get_object()
        self.perform_destroy(instance)  # 执行删除操作
        return JsonResult(data=[], code=204, msg="资源删除成功", status=status.HTTP_204_NO_CONTENT)

    def update(self, request, *args, **kwargs):
        """
        更新指定的对象，并设置更新人。
        """
        partial = kwargs.pop("partial", False)
        instance = self.get_object()
        request.data["update_by"] = request.user.id  # 设置更新人

        serializer = self.get_serializer(instance, data=request.data, partial=partial)
        serializer.is_valid(raise_exception=True)
        self.perform_update(serializer)  # 更新对象

        # 清除实例的预取缓存
        if getattr(instance, "_prefetched_objects_cache", None):
            instance._prefetched_objects_cache = {}

        return JsonResult(data=serializer.data, msg="更新成功", code=200, status=status.HTTP_200_OK)

    def list(self, request, *args, **kwargs):
        """
        列出所有对象，支持分页和过滤。
        """
        queryset = self.filter_queryset(self.get_queryset())
        page = self.paginate_queryset(queryset)

        if page is not None:
            serializer = self.get_serializer(page, many=True)
            return self.get_paginated_response(serializer.data)

        items = self.get_serializer(queryset, many=True).data
        return JsonResult(data=items, code=200, msg="获取成功", status=status.HTTP_200_OK)
        # return XopsResponse(data={"content": data, "totalElements": len(data)}, status=status.HTTP_200_OK)

    def retrieve(self, request, *args, **kwargs):
        """
        根据ID检索单个对象。
        """
        instance = self.get_object()
        serializer = self.get_serializer(instance)
        return JsonResult(data=serializer.data, code=200, msg="获取成功", status=status.HTTP_200_OK)

    def get_queryset(self):
        """
        获取对象的查询集，支持根据日期范围过滤。
        """
        query_params = self.request.query_params
        range_filters = {}

        # 处理查询参数，支持日期范围
        for item in query_params:
            if "date" in item or "time" in item:
                params = query_params.get(item, None)
                if "," in params:
                    range_filters[item + "__range"] = params.split(",")  # 添加范围过滤
                else:
                    range_filters[item] = params

        # 过滤已标记为删除的对象
        return super().get_queryset().filter(del_flag=False, **range_filters)

分页器

class CorePagination(PageNumberPagination):
    page_size = 10
    page_size_query_param = "size"


class LargePagination(PageNumberPagination):
    """自定义分页类，支持动态页面大小"""

    # 允许通过 URL 查询参数 'size' 动态设置页面大小
    page_size_query_param = "size"

    def get_paginated_response(self, data):
        """
        返回分页后的响应，包含自定义的响应格式
        """
        # 使用三元表达式简化 code 和 msg 的赋值逻辑
        code = 200 if data else 404
        msg = "success" if data else "Data Not Found"

        # 构建分页响应
        return Response(
            OrderedDict(
                [
                    ("code", code),  # 自定义状态码
                    ("msg", msg),  # 自定义消息
                    ("count", self.page.paginator.count),  # 总数据量
                    ("next", self.get_next_link()),  # 下一页链接
                    ("previous", self.get_previous_link()),  # 上一页链接
                    ("data", data),  # 实际的数据
                ]
            )
        )

统一返回

class XopsResponse(Response):
    def __init__(
        self,
        message="成功",
        status=200,
        data=None,
        template_name=None,
        headers=None,
        exception=False,
        content_type=None,
    ):

        super(Response, self).__init__(None, status=status)

        if isinstance(data, Serializer):
            message = "You passed a Serializer instance as data, but probably meant to pass serialized `.data` or `.error` representation."
            raise AssertionError(message)
        # if status >= 400:
        #     message = '失败'
        self.data = {"code": status, "message": message, "detail": data}
        self.template_name = template_name
        self.exception = exception
        self.content_type = content_type

        if headers:
            for name, value in text_type.iteritems(headers):
                self[name] = value

异常拦截

pass