Django 反向查找以获取最新版本

Question

我有一个模型结构如下，

ACTIVE_STATUS = ['waiting', 'loading', 'starting', 'running', 'stopping']
INACTIVE_STATUS = ['stopped', 'finished', 'failed', 'lost']
ALL_STATUS = ACTIVE_STATUS + INACTIVE_STATUS


class Task(models.Model):
    name = models.CharField(max_length=20)


class Job(models.Model):
    task = models.ForeignKey(Task, related_name='jobs')
    timestamp = models.DateTimeField(auto_now_add=True)
    status = models.CharField(choices=zip(ALL_STATUS, ALL_STATUS), max_length=20)

如何将“最新时间戳及其状态”注释到任务查询集中？

我已经设法获得最新的时间戳，

Task.objects.annotate(latest_ts=models.Max(models.F('job__timestamp')))

那么，怎样才能得到对应的status呢？

更新1

该查询的最大目的是对Task查询集进行排序

1. 零工作（比如说Task.objects.filter(job__isnull=True)）
2. 最新作业=='正在运行'

更新2

用于获取排序查询集的 TaskManager 类

class TaskManager(models.Manager):

    def get_queryset(self):
        qs = super().get_queryset()
        latest_job = models.Max(models.F('job__timestamp'))

        latest_status = models.Subquery(
            Job.objects.filter(
                task_id=models.OuterRef('pk')
            ).values('status').order_by('-timestamp')[:1]
        )

        qs_order = models.Case(
            models.When(job__isnull=True, then=models.Value(2)),
            models.When(latest_status='running', then=models.Value(1)),
            default=models.Value(0),
            output_field=models.IntegerField()
        )

        return qs.annotate(latest_job=latest_job, latest_status=latest_status, qs_order=qs_order).order_by('-qs_order')

Answer 1

您可以使用表达式Subquery[Django-doc]：

from django.db.models import OuterRef, Subquery

Task.objects.annotate(
    latest_status=Subquery(
        Job.objects.filter(
            task_id=OuterRef('pk')
        ).values('status').order_by('-timestamp')[:1]
    )
)

基于此，您或许还可以过滤最新状态：

from django.db.models import Q
from django.db.models import OuterRef, Subquery

Task.objects.annotate(
    latest_status=Subquery(
        Job.objects.filter(
            task_id=OuterRef('pk')
        ).values('status').order_by('-timestamp')[:1]
    )
).filter(
    Q(jobs=None) | Q(latest_status='running')
)

Job或者我们可以通过 a等的存在来排序：

from django.db.models import BooleanField, Exists, ExpressionWrapper, Max, Q
from django.db.models import OuterRef, Subquery

Task.objects.annotate(
    latest_status=Subquery(
        Job.objects.filter(
            task_id=OuterRef('pk')
        ).values('status').order_by('-timestamp')[:1]
    ),
    latest_job=Max('jobs__timestamp')
).order_by(
    Exists(Job.objects.filter(task_id=OuterRef('pk'))).asc(),
    ExpressionWrapper(Q(latest_status='running'), output_field=BooleanField()).asc(),
    'pk'
)

最终根据主键进行过滤以使排序具有确定性可能是一个好主意。