Django ORM递归查询WorkflowStep多层关联依赖步骤

模型定义 #

已定义WorkflowStep（工作流步骤）、WorkflowStepPR（前置依赖，PR为prerequisite缩写）两个模型，代码如下：

class WorkflowStep(BaseModel):
    name = models.CharField(max_length=200)
    workflow = models.ForeignKey('Workflow', on_delete=models.CASCADE)
    allowed_status = models.ManyToManyField('WorkflowStepStatus', related_name='workflow_step')
    default_status = models.ForeignKey('WorkflowStepStatus', blank=True, null=True, on_delete=models.SET_NULL)
    recheck_on_new_version = models.BooleanField(default=False)
    watch_folder = models.BooleanField(default=False)

    def __str__(self):
        return self.name

    class Meta:
        ordering = ['created_on'] 

class WorkflowStepPR(BaseModel):
    PR_for = models.ForeignKey('WorkflowStep', related_name='PR_for', on_delete=models.CASCADE)
    PR_step = models.ForeignKey('WorkflowStep', related_name='as_PR_step', on_delete=models.CASCADE)
    PR_step_status = models.ForeignKey('WorkflowStepStatus', on_delete=models.CASCADE)

    def __str__(self):
        return "{} 's PR - {}".format(self.PR_for, self.PR_for.workflow)

    class Meta:
        ordering = ['created_on']

业务需求 #

所有步骤的前置依赖通过WorkflowStepPR维护，依赖关系示意图如下：
工作流依赖关系示意图

• 箭头规则：箭头右侧步骤依赖箭头左侧步骤，例如步骤B是步骤C、D的直接前置依赖
• 查询目标：获取指定步骤（示例为步骤B）的所有直接+间接下游依赖步骤
  • 直接依赖：C、D
  • 间接依赖：依赖C的E、G，依赖D的H，依赖H的J

现有实现问题 #

当前已实现的代码仅能获取第一层直接依赖，无法拿到多层间接依赖，代码如下：

@receiver(pre_save, sender=WorkflowInstanceStep)
def status_of_workflowInstanceStep_updated(sender, instance,**kwargs):
    if instance.id is None:
        pass

    else:
        workflow_step = instance.workflow_step
        dependent_steps = workflow_step.as_PR_step.all()
        print(dependent_steps)

问题原因：as_PR_step是WorkflowStepPR中PR_step字段的反向关联，仅能查询直接关联的一层依赖，不会自动递归遍历多层关系

实现方案 #

方案1：Python层递归遍历（适合依赖层级<10层、总步骤数少的场景） #

通过BFS广度优先遍历逐层收集下游依赖，用集合自动去重避免循环依赖导致死循环：

def get_all_downstream_steps(target_step):
    collected_steps = set()
    process_queue = list(target_step.as_PR_step.all())
    
    while process_queue:
        current_step = process_queue.pop(0)
        if current_step not in collected_steps:
            collected_steps.add(current_step)
            # 将当前步骤的直接下游加入队列，继续遍历下一层
            process_queue.extend(list(current_step.as_PR_step.all()))
    
    return collected_steps

修改原有信号逻辑调用该方法即可：

@receiver(pre_save, sender=WorkflowInstanceStep)
def status_of_workflowInstanceStep_updated(sender, instance,**kwargs):
    if instance.id is None:
        return
    workflow_step = instance.workflow_step
    all_dependent_steps = get_all_downstream_steps(workflow_step)
    print(all_dependent_steps)

该方案无需额外依赖，兼容所有数据库，仅在数据量小、层级浅的场景下性能可接受。

方案2：数据库递归CTE查询（适合数据量大、依赖层级深的场景，性能更优） #

如果使用PostgreSQL、MySQL 8.0+等支持递归公用表表达式（CTE）的数据库，可以借助django-cte包在数据库层面完成递归查询，避免Python层遍历的性能损耗：

1. 安装依赖包

   pip install django-cte
   

2. 给WorkflowStepPR模型添加CTE管理器

   from django_cte import CTEManager
   
   class WorkflowStepPR(BaseModel):
       objects = CTEManager()
       # 原有字段、Meta、__str__方法保持不变
       PR_for = models.ForeignKey('WorkflowStep', related_name='PR_for', on_delete=models.CASCADE)
       PR_step = models.ForeignKey('WorkflowStep', related_name='as_PR_step', on_delete=models.CASCADE)
       PR_step_status = models.ForeignKey('WorkflowStepStatus', on_delete=models.CASCADE)
   
       def __str__(self):
           return "{} 's PR - {}".format(self.PR_for, self.PR_for.workflow)
   
       class Meta:
           ordering = ['created_on']
   

3. 实现CTE递归查询

   from django_cte import With
   
   def get_all_downstream_steps_cte(target_step):
       # 递归初始条件：第一层直接依赖当前步骤的节点
       cte = With(
           WorkflowStepPR.objects.filter(PR_step=target_step).values("PR_for_id")
       )
       # 递归逻辑：关联查询下一层依赖节点
       recursive_query = cte.join(
           WorkflowStepPR, PR_step_id=cte.col.PR_for_id
       ).values("PR_for_id")
       cte = cte.union(recursive_query, all=False)
       
       # 提取所有下游步骤ID，返回对应的WorkflowStep查询集
       downstream_ids = [item["PR_for_id"] for item in cte.queryset().with_cte(cte)]
       return WorkflowStep.objects.filter(id__in=downstream_ids)
   

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