分页器

class CorePagination(PageNumberPagination):
    """分页设置"""

    page_size = 10
    page_size_query_param = "size"
    max_page_size = 100


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

    # 定义前端分页参数名称
    page_query_params = "current"

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

    def get_paginated_response(self, data):
        """返回分页后的响应，包含自定义的响应格式"""
        # 使用三元表达式简化 code 和 msg 的赋值逻辑
        code = 200 if data else 404
        msg = "请求成功" if data else "数据请求失败"

        # TODO： 自定义ViewSet分页响应
        return Response(
            OrderedDict(
                [
                    ("code", code),
                    ("msg", msg),
                    (
                        "data",
                        OrderedDict(
                            [
                                ("records", data),
                                ("current", self.page.number),
                                ("size", self.page_size),
                                ("total", self.page.paginator.count),
                            ]
                        ),
                    ),
                ]
            )
        )

    # 动态获取前端分页参数名称
    def get_page_query_param(self):
        """
        获取 URL 上的 'current' 参数，如果没有 'current' 参数，返回 'current' 作为默认值
        :return:
        """
        return self.page_query_params or "current"

过滤器

class QueryFilter(django_filters.rest_framework.FilterSet):
    """
    自定义查询过滤器，增加常用功能
    """

    del_flag = django_filters.rest_framework.BooleanFilter(field_name="del_flag", lookup_expr="exact", required=False)
    create_date = django_filters.rest_framework.DateFromToRangeFilter(field_name="create_date", label="创建日期范围")
    update_date = django_filters.rest_framework.DateFromToRangeFilter(field_name="update_date", label="更新日期范围")
    search = django_filters.rest_framework.CharFilter(method="filter_by_all_fields", label="模糊搜索")

    class Meta:
        model = BaseEntity
        fields = ["del_flag", "create_date", "update_date"]  # 定义需要查询的字段

    def filter_queryset(self, queryset):
        """重写 filter_queryset 方法，自动过滤 del_flag=True 的数据"""
        queryset = queryset.filter(del_flag=False)
        queryset = super().filter_queryset(queryset)
        return queryset

    def filter_by_all_fields(self, queryset, name, value):
        """模糊搜索，搜索所有文本字段"""
        for field in self.Meta.model._meta.fields:
            if isinstance(field, (CharField, TextField)):
                queryset = queryset.filter(**{f"{field.name}__icontains": value})
        return queryset

权限类

class RbacPermission(BasePermission):
    """
    自定义权限类，基于用户角色权限控制。
    """

    @classmethod
    def get_permission_from_role(cls, request) -> Optional[List[str]]:
        """
        获取用户角色的权限，去重
        :param request: 请求对象
        :return: 权限列表或 None
        """
        if not request.user.is_authenticated:
            return None
        return list(request.user.roles.values_list("perms__method", flat=True).distinct())

    def has_permission(self, request, view) -> bool:
        """
        权限校验逻辑
        :param request: 请求对象
        :param view: 视图对象
        :return: 是否有权限
        """
        perms = self.get_permission_from_role(request) or []

        if "admin" in perms:
            return True  # 管理员直接拥有所有权限

        # 如果视图没有权限映射，直接返回有权限
        if not hasattr(view, "perms_map"):
            return True

        # 获取请求方法
        method = request.method.lower()

        # 校验权限
        return any(
            (method == method_key or method_key == "*") and alias in perms
            for perm_map in view.perms_map
            for method_key, alias in perm_map.items()
        )




class ObjPermission(BasePermission):
    """密码管理对象级权限控制"""

    def has_object_permission(self, request, view, obj) -> bool:
        """
        检查用户是否有权限访问特定对象
        :param request: 请求对象
        :param view: 视图对象
        :param obj: 要检查的对象
        :return: 是否有权限
        """
        # 获取用户角色的权限
        user_perms = RbacPermission.get_permission_from_role(request) or []

        # 管理员拥有所有权限
        if "admin" in user_perms:
            return True

        # 检查用户是否有权限访问特定对象
        return request.user.id == obj.uid_id