我有一个模型结构如下,
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
查询集进行排序
- 零作业(比如
Task.objects.filter(job__isnull=True)
) - 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')
最佳答案
您可以使用 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'
)
最终过滤主键以使排序确定性可能是个好主意。
关于python - Django 反向查找获取最新,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/62801831/