如何在Django QuerySet中动态添加自定义分组？

问题描述 #

我有Report和ReportGroup模型，二者为多对多关系。目前可正常获取所有ReportGroup实例的QuerySet，结构如下：

Group Name 1:
    Report Name 1
    Report Name 2
  Group Name 2:
    Report Name 2
    Report Name 10
  ... and etc.

现在需要在该QuerySet中添加动态生成的"Total Reports"分组，结构如下：

Group All reports:
    Report Name 1
    Report Name 2
    Report Name 5
  ... and other reports from QuerySet[Report]

限制条件：

• 由于QuerySet受对象权限控制，无法将"Total Reports"分组存入数据库，必须动态生成
• Serializer要求传入合法的QuerySet

尝试过的代码 #

我曾尝试以下写法，但调用total.reports.set()时出现错误：needs to have a value for field 'id'...

def get_queryset(self):
  groups = get_objects_for_user(
    user=self.request.user,
    perms=view_reportgroup,
    klass=ReportGroup.actived_objects,
    accept_global_perms=False
  )
  reports = get_objects_for_user(
    user=self.request.user,
    perms=view_report,
    klass=Report.actived_objects,
    accept_global_perms=False
  )
  total = ReportStorage(id=None, name='Total Reports')
  total.reports.set(
    Report.actived_objects.filter(
      Q(id__in=reports) | Q(groups__in=groups)
    ).distinct()
  )
  return (groups | total).prefetch_related(
    Prefetch(
      'reports',
      queryset=ReportStorage.prefetch_related(
        Prefetch(
          'reportstorage',
          queryset=super().get_queryset()
        )
      )
    )
  )

我还考虑过将ReportStorage实例添加到groups._result_cache，但不清楚具体实现。也想过自定义SQL或者把QuerySet转为列表，想知道有没有更合理的方案。

现有ViewSets代码 #

class ReportStorageGenericViewSet(GenericViewSet):
    """
    Basic view for filtering file storage objects.

    Fields
    -------
    queryset: QuerySet[ReportStorage]
        Basic set of objects.
    query_params_validator: QueryParamValidator
        Class for processing the correctness of the specified date from query_param.
    """
    queryset = ReportStorage.objects
    query_params_validator = QueryParamsValidator()  # getattr(settings, 'DATE_INPUT_FORMATS', None))
    permission_classes = [PortalRDjangoObjectPermissions]

    def _get_date(self) -> date:
        """
        Getting date from query_params.

        :return: Valid date, if the value is not specified, the reporting day is used.
        """
        date_str = self.request.query_params.get('date')
        self.query_params_validator(date_str)
        if date_str:
            return parse_date(date_str)

        return (timezone.now() - timedelta(days=1)).date()

    def get_queryset(self) -> QuerySet[ReportStorage]:
        """
        Generating a list of objects.

        It is guaranteed that the list of files is filtered by date.

        :return: A set of ReportStorage objects.
        """
        return super().get_queryset().filter(validity_period__contains=self._get_date())

    def get_serializer_context(self) -> Dict[str, Any]:
        """
        Adding a queryset to the context during the List operation to further check for the presence of loaded help at the current time.

        :return: Context for the serializer.
        """
        context = super().get_serializer_context()
        if self.action == 'list':
            context['queryset'] = self.get_queryset()

        return context

    def handle_exception(self, exc: APIException) -> Response:
        """
        Handling request execution errors.

        :param exc: Query execution exception.
        :return: If an error occurs, returns a standard {"detail": "message"} JSON response.
        """
        if isinstance(exc, ValidationError):
            return Response({"detail": exc.detail}, status=exc.status_code)

        return super().handle_exception(exc=exc)

class ReportGroupViewSet(ListModelMixin, ReportStorageGenericViewSet):
    """
    View to get a list of reporting groups with enabled reports and file uploads.

    Fields
    -------
    serializer_class: ReportGroupSerializer
        Basic view object serializer.
    """
    serializer_class = ReportGroupSerializer

    def get_queryset(self) -> QuerySet[ReportGroup]:
        """
        Generating a list of objects.

        :return: List of ReportGroup objects.
        """
        #### Override queryset here ####
        return ReportGroup.actived_objects.prefetch_related(
            Prefetch(
                'reports',
                queryset=Report.actived_objects.prefetch_related(
                    Prefetch('storage', queryset=super().get_queryset())
                )
            )
        )

解决方案建议 #

方案1：将QuerySet转为列表后添加动态实例 #

这是最直接的实现方式，无需修改Serializer核心逻辑，仅需在get_queryset中将QuerySet转为列表，手动添加动态生成的分组实例：

def get_queryset(self) -> list[ReportGroup]:
    # 获取权限过滤后的分组，预先完成关联数据预取
    groups = get_objects_for_user(
        user=self.request.user,
        perms=view_reportgroup,
        klass=ReportGroup.actived_objects,
        accept_global_perms=False
    ).prefetch_related(
        Prefetch(
            'reports',
            queryset=Report.actived_objects.prefetch_related(
                Prefetch('storage', queryset=super().get_queryset())
            )
        )
    )
    # 获取用户有权限查看的所有报告，同样完成预取
    allowed_reports = get_objects_for_user(
        user=self.request.user,
        perms=view_report,
        klass=Report.actived_objects.prefetch_related(
            Prefetch('storage', queryset=super().get_queryset())
        ),
        accept_global_perms=False
    ).distinct()
    
    # 动态生成Total Reports分组，用负ID区分数据库对象
    total_group = ReportGroup(id=-1, name='Total Reports')
    # 手动绑定reports属性，避免调用set()触发未保存对象的数据库操作
    total_group.reports = allowed_reports
    
    # 转为列表后添加动态分组
    groups_list = list(groups)
    groups_list.append(total_group)
    
    return groups_list

注意：需确保ReportGroupSerializer能处理未保存的实例，只要序列化name和reports字段即可，无需验证数据库存在性。

方案2：在响应层面追加动态数据 #

如果不需要返回严格的QuerySet，可直接在list方法中先序列化原分组数据，再追加"Total Reports"条目，对原逻辑侵入最小：

def list(self, request, *args, **kwargs):
    # 获取并序列化原分组数据
    queryset = self.filter_queryset(self.get_queryset())
    serializer = self.get_serializer(queryset, many=True)
    response_data = serializer.data
    
    # 获取并序列化用户有权限的所有报告
    allowed_reports = get_objects_for_user(
        user=self.request.user,
        perms=view_report,
        klass=Report.actived_objects.prefetch_related(
            Prefetch('storage', queryset=super().get_queryset())
        ),
        accept_global_perms=False
    )
    report_serializer = ReportSerializer(allowed_reports, many=True, context=self.get_serializer_context())
    
    # 追加Total Reports条目
    response_data.append({
        'id': -1,
        'name': 'Total Reports',
        'reports': report_serializer.data
    })
    
    return Response(response_data)

方案3：自定义SQL构造Union查询（复杂场景适用） #

如果必须返回严格的QuerySet，可通过自定义SQL构造虚拟行与原QuerySet做Union，但需要手动处理多对多关联的预取逻辑，实现复杂度较高，仅适合对QuerySet有强依赖的场景。

内容的提问来源于stack exchange，提问作者Qngdjas