DRFmixin随机返回queryset

• Z时代
• 2024-01-10
• 分类：综合

mixin.py(自己封装，方便重用)

class RandomMixin:
    page_size = 15
    @classmethod
    def filter_selected_pks(cls, queryset=None, selected_pks=None):
        queryset = queryset.filter(pk__in=selected_pks)
        return queryset
    @classmethod
    def random_pk_list(cls, queryset=None, page_size=15):
        """
        :param queryset:
        :param page_size:
        :return:
        """
        if not isinstance(queryset, QuerySet):
            raise TypeError("queryset must be QuerySet instance")
        pk_list = queryset.values_list("pk", flat=True)
        logger.debug(f"page_size:{page_size} pk_list:{len(pk_list)}")
        if len(pk_list) < page_size:
            selected_len = len(pk_list)
        else:
            selected_len = page_size
        selected_pks = random.sample(list(pk_list), selected_len)
        logger.info(f"selected_pks:{selected_pks}")
        return selected_pks
    @classmethod
    def random_queryset(cls, queryset=None, page_size=15):
        selected_pks = cls.random_pk_list(queryset, page_size=page_size)
        queryset = cls.filter_selected_pks(queryset, selected_pks=selected_pks)
        return selected_pks, queryset
    @classmethod
    def random_queryset_data(cls, sort_keys=None, serializer_data=None):
        return_dict = OrderedDict().fromkeys(sort_keys)
        for item in serializer_data:
            return_dict[item.get("id")] = item
        return sort_keys, list(return_dict.values())
    def get_random_queryset(self, validated_data=None):
        """
        :param validated_data:
        :return:
        """
        raise NotImplementedError(
            "`get_random_queryset()` must be implemented.")
    def list_random(self, request, *args, **kwargs):
        """
        随机列表
        ----
        """
        logger.info("__class__:{} list_random request.data:{}".format(
            self.__class__, request.data))
        queryset = self.get_random_queryset()
        selected_pks, random_queryset = self.random_queryset(
            queryset=queryset, page_size=self.random_activity_size,
        )
        serializer = self.get_serializer(random_queryset, many=True)
        _, data = self.random_queryset_data(
            sort_keys=selected_pks,
            serializer_data=serializer.data,
        )
        logger.info("random_queryset_data data len:{}".format(len(data)))
        return Response(make_response(data=data))

views.py 调用方继承RandomMixin, 实现抽象方法.

class ActivityInfoSet(mixins.ListModelMixin, mixins.RetrieveModelMixin,
                      RandomMixin,
                      AuthorizationStatusMixin,
                      AllowAnyViewSet):
    """
    ActivityInfo Set
    """
    queryset = ActivityInfo.objects.all() 
        .filter(is_deleted=False)
        .select_related("category_type")
    serializer_class = ActivityInfoSerializer
    filterset_class = ActivityInfoTypeFilter
    page_size = settings.REST_FRAMEWORK.get("PAGE_SIZE")
    random_activity_size = 6
    def get_queryset(self):
        return self.get_pass_queryset()
    @swagger_auto_schema(responses={200: ActivityInfoSerializer(many=True)})
    def list(self, request, *args, **kwargs):
        """
        活动列表
        ----
        """
        return super().list(request, *args, **kwargs)
    def get_random_queryset(self, validated_data=None):
        queryset = self.filter_queryset(self.get_queryset())
        return queryset
    @swagger_auto_schema(responses={200: ActivityInfoSerializer(many=True)})
    @action(detail=False, methods=["GET"])
    def list_random(self, request, *args, **kwargs):
        """
        活动随机列表
        ----
        """
        return super().list_random(request, *args, **kwargs)
    @swagger_auto_schema(responses={200: ActivityInfoSerializer})
    def retrieve(self, request, *args, **kwargs):
        """
        活动详情
        ----
        """
        return super().retrieve(request, *args, **kwargs)

以上是 DRFmixin随机返回queryset 的全部内容， 来源链接： utcz.com/z/516371.html