查询所有行并返回每个重复的最新行

Posted 2023-02-24

【中文标题】查询所有行并返回每个重复的最新行

【英文标题】：Query all rows and return most recent of each duplicate

【发布时间】：2011-04-29 20:37:41

【问题描述】：

我有一个模型，它的 id 不是唯一的。每个模型也有一个日期。我想返回所有结果，但只返回共享 id 的每一行的最新结果。模型看起来像这样：

class MyModel(models.Model):
    my_id = models.PositiveIntegerField()
    date  = models.DateTimeField()
    title = models.CharField(max_length=36)


## Add some entries
m1 = MyModel(my_id=1, date=yesterday, title='stop')
m1.save()

m2 = MyModel(my_id=1, date=today, title='go')
m2.save()

m3 = MyModel(my_id=2, date=today, title='hello')
m3.save()

现在尝试检索这些结果：

MyModel.objects.all()... # then limit duplicate my_id's by most recent

结果应该只有 m2 和 m3

【参考方案1】：

仅使用 ORM 无法做到这一点，您需要获取所有记录，然后在 Python 中丢弃重复项。

例如：

objs = MyModel.objects.all().order_by("-date")
seen = set()
keep = []
for o in objs:
    if o.id not in seen:
        keep.append(o)
        seen.add(o.id)

这里有一些自定义的 SQL 可以从数据库中得到你想要的：

select * from mymodel where (id, date) in (select id, max(date) from mymodel group by id)

您应该能够调整它以在 ORM 中使用。

【讨论】：

一旦你循环结果，这不会评估 QuerySet 并导致所有查找吗？真的没有办法吗？

关系数据库（以及基于它们构建的 ORM）不擅长行之间的操作（包括比较）。他们的模型基本上是关于选择一组行，然后对它们进行排序。我想不出一种方法让 SQL 做你想做的事..

好的，感谢您抽出宝贵时间。我想我会以其他方式限制结果（比如只获得最近的结果）以减轻重量。再次感谢内德！

等待：我在答案中添加了一些自定义 SQL。

【参考方案2】：

您还应该考虑将上述逻辑抽象为管理器：

http://docs.djangoproject.com/en/dev/topics/db/managers/

这样你可以调用类似 MyModel.objects.no_dupes() 的东西，你可以在管理器中定义 no_dupes() 并执行 Ned 在那里布置的逻辑。

您的 models.py 现在看起来像这样：

class MyModelManager(models.Manager):
    def no_dupes:
        objs = MyModel.objects.all().order_by("-date")
        seen = set()
        keep = []
        for o in objs:
            if o.id not in seen:
                keep.append(o)
                seen.add(o.id)
        return keep

class MyModel(models.Model):
    my_id = models.PositiveIntegerField()
    date  = models.DateTimeField()
    title = models.CharField(max_length=36)
    objects = MyModelManager()

有了上面的代码，你可以调用：MyModel.objects.no_dupes()，这应该会得到你想要的结果。如果您愿意，您甚至可以覆盖 all() 函数：

http://docs.djangoproject.com/en/1.2/topics/db/managers/#modifying-initial-manager-querysets

如果您需要在整个项目的多个视图中使用它，我发现管理器是一个更好的解决方案，这样您就不必重写代码 X 次。

【讨论】：

无论我是想把过滤放在自定义管理器中还是视图中，难道我还需要获取所有记录然后过滤它们吗？如果可能的话，我真的很想在进行实际的数据库调用之前进行过滤。这可能吗？

您可以使用管理器修改实际的 SQL 查询。看这里的例子：docs.djangoproject.com/en/dev/topics/db/managers/…

感谢有关模型管理器的提示，我没有考虑过制作自定义管理器。

【参考方案3】：

正如 Ned 所说，我不知道使用 ORM 的方法。但是你也许可以使用 db 来限制你必须在 python 的 for 循环中做的工作量。

这个想法是使用 Django 的 annotate（它基本上是在运行 group_by）来查找所有具有多个具有相同 my_id 的行的实例，并按照 Ned 的建议处理它们。然后对于其余的（没有重复的），您可以只获取各个行。

from django.db.models import Count, Q
annotated_qs = MyModel.objects.annotate(num_my_ids=Count('my_id')).order_by('-date')
dupes = annotated_qs.filter(num_my_ids__gt=1)
uniques = annotated_qs.filter(num_my_ids__lte=1)
for dupe in dupes:
   ... # just keep the most recent, as Ned describes
keep_ids = [keep.id for keep in keeps]
latests = MyModel.objects.filter(Q(id__in=keep_ids) | Q(id__in=uniques))

如果你只有少量的欺骗，这意味着你的 for 循环要短得多，代价是额外的查询（获取欺骗）。