python - 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. latest_job=='正在运行'

更新-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 expression [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 等的存在来排序:

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

最终过滤主键以使排序确定性可能是个好主意。