如何构造复杂的 django 查询语句？答案

【问题标题】：How do I construct complex django query statements?如何构造复杂的 django 查询语句？
【发布时间】：2020-11-04 09:56:29
【问题描述】：

我对 SQL 不是很熟悉，所以试图通过 Django ORM 进行更复杂的调用让我很困惑。我有一个生成 Jobs 的 Printer 模型，并且这些 Jobs 通过一个 State 模型接收状态，它与它有外键关系。作业状态由与其关联的最新状态对象确定。这样我就可以在整个生命周期中跟踪作业状态的历史。我希望能够确定哪些打印机具有与之关联的成功作业。

from django.db import models


class Printer(models.Model):
    label = models.CharField(max_length=120) 


class Job(models.Model):
    label = models.CharField(max_length=120)
    printer = models.ForeignKey(
        Printer,
        related_name='jobs',
        related_query_name='job'
    )
    
    def set_state(self, state):
        State.objects.create(state=state, job=self)

    @property
    def current_state(self):
        return self.states.latest('created_at').state


class State(models.Model):
    created_at = models.DateTimeField(auto_now_add=True)
    state = models.SmallIntegerField()
    job = models.ForeignKey(
        Job,
        related_name='states',                                          
        related_query_name='state'
    )

我需要一个打印机对象的 QuerySet，这些对象至少有一个相关作业，其最近（最新）状态对象具有 State.state == '200'。有没有办法构造一个复合调用，它可以使用数据库来实现这一点，而不必拉入所有 Job 对象来运行 python 迭代？也许是定制经理？我一直在阅读有关 Subquery、Annotation 和 OuterRef 的帖子，但这些想法并没有以一种向我展示路径的方式深入人心。我需要像我 5 岁那样解释它们。它们是非常不合 Python 的语句..

描述我想要的天真的python方式：

printers = []
for printer in Printer.objects.all():
    for job in printer.jobs.objects.all():
        if job.states.latest().state == '200':
            printers.append(printer)
printers = list(set(printers))

但尽可能减少数据库往返次数。救命！

编辑：进一步的问题，根据当前状态过滤作业的最佳方法是什么。由于 Job.current_state 是一个计算属性，它不能在 QuerySet 过滤器中使用。但是，我不想再拉入所有 Job 对象。

【问题讨论】：

标签： sql django django-orm

【解决方案1】：

花了大约两天时间，但我想我使用注释和子查询有一个答案：

state_sq = State.objects.filter(job=OuterRef('pk')).order_by('-created_at')                                                                        

successful_jobs = Job.objects.annotate(          
    latest_state=Subquery(state_sq.values('state')[:1])        
).filter(printer=OuterRef('pk'), latest_state='200')
                                                                        
printers_with_successful_jobs = Printer.objects.annotate(                                   
    has_success_jobs=Exists(successful_jobs)                           
).filter(has_success_jobs=True)

此外，我构建了一个自定义管理器，默认返回latest_state。

class JobManager(models.Manager):                                       
    def get_queryset(self):                                             
        state_sq = State.objects.filter(                                
            object_id=OuterRef('pk')                                    
        ).order_by('-created_at')                                       
                                                                        
        return super().get_queryset().annotate(                         
            latest_state=Subquery(state_sq.values('state')[:1])         
        )

class Job(models.Model):
    objects = JobManager()
    ...

【讨论】：