如何构造复杂的django查询语句?

o2gm4chl  于 2021-07-26  发布在  Java
关注(0)|答案(1)|浏览(271)

我对sql不是很熟悉,所以试图通过django orm进行更复杂的调用让我很为难。我有一个生成作业的打印机模型,作业通过一个状态模型接收状态,该状态模型具有外键关系。作业状态由与其关联的最新状态对象确定。这样我就可以在工作的整个生命周期中跟踪工作状态的历史。我想能够确定哪些打印机有成功的工作与他们有关。

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'
    )

我需要一个打印机对象的查询集,其中至少有一个与其最近(最新)状态对象相关的作业 State.state == '200' . 有没有一种方法可以构造一个复合调用,使用数据库来实现这一点,而不必拉入所有作业对象来运行python迭代?也许是定制经理?我一直在读关于子查询、注解和outerref的帖子,但这些想法并没有以某种方式向我展示一条道路。我需要他们像我5岁一样解释。他们是非常不讲情话的。。
用朴素的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))

但是尽可能减少db往返次数。救命啊!
编辑:还有一个问题,根据当前状态过滤作业的最佳方法是什么。由于job.current\u state是计算属性,因此不能在queryset筛选器中使用。但是,我也不想把所有的工作对象都拉进来。

goqiplq2

goqiplq21#

花了大约两天的时间,但我想我已经用注解和子查询找到了答案:

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()
    ...

相关问题