Django中RawQuerySet无法用django-filter处理，如何转换为可用查询集？

报错信息 #

AttributeError at /tracker/
'RawQuerySet' object has no attribute 'all'

问题原因 #

使用django-filter处理原生SQL查询得到的RawQuerySet时，由于RawQuerySet不支持Django ORM的all()方法（而django-filter依赖该方法实现过滤逻辑），导致触发上述报错。

问题代码 #

def track1(request):
    sql = """
        select 1 as id,
            p.name project,
            i.title issue,
            u.name,
            replace(ROUND(t.time_spent/3600.0, 1)::text, '.', ',') as spent,
            TO_CHAR(t.spent_at + interval '2h', 'dd.mm.yyyy HH24:MI:SS') date_spent, substring(n.note for 300) note
        from issues i
        left join projects p on p.id = i.project_id
        left join timelogs t on t.issue_id = i.id
        left join users u on u.id = t.user_id
        left join notes n on n.id = t.note_id
        where (t.spent_at + interval '2h') between '2022-06-01' and '2022-06-30 23:59:59'
        order by 5, 1, 2
    """

    user_spent_on_project = UsersSpentOnProjects.objects.raw(sql)
    filter = UsersSpentOnProjectsFilter(request.GET, queryset=user_spent_on_project)
    user_spent_on_project = filter.qs

    context = {
        'user_spent_on_project' : user_spent_on_project,
        'filter' : filter
    }

    return render(request, 'trackApp/track1.html', context=context)

提问 #

有没有办法将这个RawQuerySet转换为类似Model.objects.all()的可被过滤的查询集？

解决方案 #

方法1：用Django ORM重写原生SQL（最优解） #

将原生SQL逻辑转换为Django ORM查询，得到真正的QuerySet，直接兼容django-filter。示例代码如下（假设模型关联关系正确）：

from django.db.models import F, Value, IntegerField, CharField
from django.db.models.functions import Round, Substr, Func
from django.utils import timezone
from datetime import datetime

def track1(request):
    # 转换时间条件：原SQL中t.spent_at + 2h在指定区间，等价于t.spent_at在[2022-05-31 22:00:00, 2022-06-30 21:59:59]
    start_date = timezone.make_aware(datetime(2022, 5, 31, 22, 0, 0))
    end_date = timezone.make_aware(datetime(2022, 6, 30, 21, 59, 59))

    # 用ORM构建查询
    queryset = Timelogs.objects.filter(
        spent_at__gte=start_date,
        spent_at__lte=end_date
    ).annotate(
        id=Value(1, output_field=IntegerField()),
        project=F('issue__project__name'),
        issue=F('issue__title'),
        name=F('user__name'),
        # 实现replace(ROUND(t.time_spent/3600.0,1)::text, '.', ',')
        spent=Func(
            Round(F('time_spent')/3600.0, 1),
            Value('.'), Value(','),
            function='replace',
            output_field=CharField()
        ),
        date_spent=Func(
            F('spent_at') + Value('2 hours'),
            Value('dd.mm.yyyy HH24:MI:SS'),
            function='TO_CHAR',
            output_field=CharField()
        ),
        note=Substr('note__note', 1, 300)
    ).select_related('issue__project', 'user', 'note').order_by('spent', 'id', 'project')

    # 直接使用django-filter
    filter = UsersSpentOnProjectsFilter(request.GET, queryset=queryset)

    context = {
        'user_spent_on_project': filter.qs,
        'filter': filter
    }

    return render(request, 'trackApp/track1.html', context=context)

方法2：将RawQuerySet转换为列表（仅适用于小数据量） #

如果数据量不大，可以将RawQuerySet转换为列表，然后手动处理过滤逻辑：

user_spent_on_project = list(UsersSpentOnProjects.objects.raw(sql))

# 示例：根据request.GET中的project参数过滤
filtered_data = [item for item in user_spent_on_project if item.project == request.GET.get('project', '')]

这种方法会一次性加载所有数据到内存，无法利用数据库的过滤优化，仅适合小数据集场景。

方法3：自定义FilterSet适配RawQuerySet #

如果必须保留原生SQL，可以重写django-filter的FilterSet，手动实现过滤逻辑：

import django_filters

class UsersSpentOnProjectsFilter(django_filters.FilterSet):
    project = django_filters.CharFilter(method='filter_project')

    class Meta:
        model = UsersSpentOnProjects
        fields = ['project']

    def filter_project(self, queryset, name, value):
        # 手动遍历RawQuerySet筛选符合条件的项
        return [item for item in queryset if item.project == value]

该方法同样需要遍历所有数据，性能较差，不推荐用于大数据量场景。

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