如何在Django中获取版本化调度模型的最新已发布条目？

问题背景 #

我开发的应用包含一个调度系统，用于为工人分配不同时间段的工作地点，调度通过ChangeSet模型实现版本控制，每个ScheduleEntry关联一个ChangeSet。

模型定义 #

class Worker(models.Model):
    name = models.CharField(max_length=100)

class Location(models.Model):
    name = models.CharField(max_length=100)

class TimeBlock(models.Model):
    start_date = models.DateField()
    end_date = models.DateField()

class ChangeSet(models.Model):
    published_on = models.DateTimeField(null=True)

class ScheduleEntry(models.Model):
    change_set = models.ForeignKey(ChangeSet, on_delete=models.CASCADE)
    worker = models.ForeignKey(Worker, on_delete=models.CASCADE)
    time_block = models.ForeignKey(TimeBlock, on_delete=models.CASCADE)
    location = models.ForeignKey(Location, null=True, on_delete=models.CASCADE)

核心规则 #

每个工人在每个时间段只能被分配一个地点，由该工人和时间段对应的、拥有**最新非空change_set.published_on**的ScheduleEntry决定。工人未调度的情况包括：

• 无对应时间段的条目
• 所有条目关联未发布的变更集
• 最新已发布条目location为null

单个工人单个时间段的查询逻辑很简单：

ScheduleEntry.objects.filter(
    change_set__published_on__isnull=False,
    worker=my_worker,
    time_block=my_time_block,
).order_by('-change_set__published_on').first()

但需要实现以下跨场景查询（需兼容SQLite，不能依赖PostgreSQL的多字段distinct特性）：

• 某工人全时间段的完整调度
• 特定时间段内被分配到特定地点的所有工人

本质是要查询所有属于已发布变更集、且未被同工人/时间段对的后续（change_set.published_on更新）条目覆盖的ScheduleEntry。

解决方案 #

1. SQL原生查询思路 #

通过子查询找到每个(worker, time_block)对对应的最大published_on值，再关联原表筛选出匹配的条目：

SELECT se.*
FROM scheduleentry se
JOIN changeset cs ON se.change_set_id = cs.id
WHERE cs.published_on IS NOT NULL
AND (se.worker_id, se.time_block_id, cs.published_on) IN (
    SELECT se_inner.worker_id, se_inner.time_block_id, MAX(cs_inner.published_on)
    FROM scheduleentry se_inner
    JOIN changeset cs_inner ON se_inner.change_set_id = cs_inner.id
    WHERE cs_inner.published_on IS NOT NULL
    GROUP BY se_inner.worker_id, se_inner.time_block_id
)

2. Django ORM实现 #

利用Subquery和OuterRef实现上述逻辑，完全兼容SQLite：

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

# 子查询：获取每个(worker, time_block)对的最新已发布时间
latest_published_subquery = ScheduleEntry.objects.filter(
    worker=OuterRef('worker'),
    time_block=OuterRef('time_block'),
    change_set__published_on__isnull=False
).annotate(
    latest_published=Max('change_set__published_on')
).values('latest_published')[:1]

# 主查询：筛选出匹配最新发布时间的已调度条目
latest_schedule_entries = ScheduleEntry.objects.filter(
    change_set__published_on__isnull=False,
    change_set__published_on=Subquery(latest_published_subquery)
)

基于结果的过滤示例 #

某工人全时间段的完整调度

worker_full_schedule = latest_schedule_entries.filter(worker=my_worker)

特定时间段内被分配到特定地点的所有工人

target_workers = latest_schedule_entries.filter(
    time_block__start_date__lte=target_end_date,
    time_block__end_date__gte=target_start_date,
    location=my_location
).values_list('worker', flat=True).distinct()

内容的提问来源于stack exchange，提问作者hashemi