使用django-cte遇AttributeError：QuerySet无with_cte属性，求解决

我有Model1模型数据，想要按created字段的月份分组，获取每个月的最新记录。使用django-cte编写代码时遇到错误：'QuerySet' object has no attribute 'with_cte'，相关信息如下：

数据格式 #

| id  | name |       created                  |
-----------------------------------------------
|1    | A    |2024-04-10T02:49:47.327583-07:00|
|2    | A    |2024-04-01T02:49:47.327583-07:00|
|3    | A    |2024-03-01T02:49:47.327583-07:00|
|4    | A    |2024-02-01T02:49:47.327583-07:00|
|5    | B    |2024-02-01T02:49:47.327583-07:00|

模型定义 #

class Model1(models.Model):
    name = models.CharField(max_length=100)
    created = models.DateTimeField(auto_now_add=True)

（注：原代码中model.Models应为models.Model，已修正）

期望输出 #

| id  | name |       created                  |
-----------------------------------------------
|1    | A    |2024-04-10T02:49:47.327583-07:00|
|3    | A    |2024-03-01T02:49:47.327583-07:00|
|4    | A    |2024-02-01T02:49:47.327583-07:00|

出错代码 #

from django.db.models.functions import DenseRank, ExtractMonth
from django_cte import With
from django.db.models import F, Window

m = Model1.objects.get(id=1)
cte = With(
        Model1.objects.filter(name=m.name)
          rank=Window(
            expression=DenseRank(),
            partition_by=[ExtractMonth("created")],
            order_by=F("created").desc(),
        )
     )
qs = cte.queryset().with_cte(cte).filter(rank=1)

错误信息 #

qs = cte.queryset().with_cte(cte).filter(rank=1)
                    ^^^^^^^^^^^^^^^^^^^^^^^
AttributeError: 'QuerySet' object has no attribute 'with_cte'

问题原因 #

1. 字段添加方式错误：给QuerySet添加rank计算字段必须用.annotate()方法，原代码直接在QuerySet后写rank=属于语法错误。
2. with_cte调用位置错误：with_cte是With对象提供的方法，不能直接调用在普通QuerySet上，正确用法是将CTE绑定到最终的查询链中。

修正后的代码 #

from django.db.models.functions import DenseRank, ExtractMonth
from django_cte import With
from django.db.models import F, Window

m = Model1.objects.get(id=1)

# 定义CTE：按月份分组，给每组记录按created降序排名
cte = With(
    Model1.objects.filter(name=m.name)
    .annotate(
        rank=Window(
            expression=DenseRank(),
            partition_by=[ExtractMonth("created")],
            order_by=F("created").desc(),
        )
    )
)

# 基于CTE筛选排名为1的记录（即每个月的最新记录）
qs = cte.queryset().filter(rank=1).with_cte(cte)

# 执行查询并查看结果
for item in qs:
    print(item.id, item.name, item.created)

关键说明 #

• 使用.annotate()是Django给QuerySet添加计算字段的标准方式，必须通过这个方法来注入rank窗口函数。
• with_cte(cte)需要调用在最终的QuerySet上，用来告知Django执行查询时先处理CTE定义。
• 原模型定义中的model.Models是笔误，正确应为models.Model，否则模型无法正常创建。

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