如何在Django信号中编写原生SQL CTE获取员工层级结构

有一个团队层级模型代码如下：

class Hrc(models.Model):
    id = models.UUIDField(default=uuid.uuid4, primary_key=True, unique=True, editable=False)
    emp = models.ForeignKey('Employee', on_delete=models.CASCADE, related_name='emp')
    mang = models.ForeignKey('Employee', on_delete=models.CASCADE, related_name='mang')
    created_on = models.DateTimeField(auto_now_add=True)

    class Meta:
        ordering = ['created_on']

需要在Employee的post_save信号中，当员工信息更新时，用CTE获取该员工的下属层级结构，目标CTE的SQL逻辑如下：

with hierarchy as 
(
  select emp, mang, 1 as lvl
  from hrc 
  where emp = instance
  
  union all
  
  select child.emp, parent.emp, lvl+1
  from hierarchy as parent
  join hrc as child
      on child.mang = parent.emp
), lvls as (
    select lvl, emp
    from hierarchy
    group by lvl, emp
)
select lvl, STRING_AGG(emp, '') within group (order by emp)
from lvls
group by lvl;

请教如何在信号函数中实现这个CTE并得到预期结果？

前置修正 #

原CTE的初始条件逻辑有误：where emp = instance是查询当前员工作为下属的记录，而我们需要获取当前员工的下属，所以应改为where mang = instance（即找直接汇报给该员工的下属），后续递归逻辑也需对应调整。

下面提供两种实现方式：

方式一：使用原生SQL执行CTE #

适合需要完全复用原生SQL逻辑的场景，以下示例基于PostgreSQL，其他数据库需调整聚合函数：

from django.db import connection
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Employee

@receiver(post_save, sender=Employee)
def employeeUpdated(sender, instance, created, **kwargs):
    if not created:
        with connection.cursor() as cursor:
            # 修正后的CTE SQL，用参数占位符避免SQL注入
            sql = """
            WITH hierarchy AS (
              SELECT emp_id, mang_id, 1 AS lvl
              FROM hrc 
              WHERE mang_id = %s
              
              UNION ALL
              
              SELECT child.emp_id, parent.emp_id, parent.lvl + 1
              FROM hierarchy AS parent
              JOIN hrc AS child ON child.mang_id = parent.emp_id
            ), lvls AS (
                SELECT lvl, emp_id
                FROM hierarchy
                GROUP BY lvl, emp_id
            )
            SELECT lvl, STRING_AGG(CAST(emp_id AS TEXT), ',' ORDER BY emp_id)
            FROM lvls
            GROUP BY lvl;
            """
            # 传递当前员工ID作为参数
            cursor.execute(sql, [instance.id])
            # 获取并转换结果格式
            results = cursor.fetchall()
            hierarchy_data = [{"层级": row[0], "员工ID列表": row[1]} for row in results]
            # 此处可添加后续业务逻辑，比如保存结果或触发其他操作
            print(hierarchy_data)

方式二：使用Django ORM的CTE API（推荐） #

Django 3.2+支持原生CTE的ORM封装，更符合Django开发风格，自动处理SQL注入和数据库适配：

from django.db import models
from django.db.models import F, Value
from django.db.models.expressions import CTE
from django.db.models.functions import Cast, StringAgg
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Employee, Hrc

@receiver(post_save, sender=Employee)
def employeeUpdated(sender, instance, created, **kwargs):
    if not created:
        # 初始CTE：获取当前员工的直接下属，层级设为1
        hierarchy_cte = CTE(
            Hrc.objects.filter(mang=instance).annotate(
                lvl=Value(1, output_field=models.IntegerField())
            ).values('emp_id', 'mang_id', 'lvl')
        )
        # 递归拼接：获取下属的下属，层级递增
        hierarchy_cte = hierarchy_cte.union_all(
            Hrc.objects.filter(
                mang_id=hierarchy_cte.col.emp_id
            ).annotate(
                lvl=hierarchy_cte.col.lvl + 1
            ).values('emp_id', 'mang_id', 'lvl')
        )
        # 按层级聚合员工ID，用逗号分隔并排序
        result_query = hierarchy_cte.queryset.values('lvl').annotate(
            员工ID列表=StringAgg(
                Cast('emp_id', output_field=models.CharField()),
                separator=',',
                ordering=F('emp_id')
            )
        ).order_by('lvl')
        
        # 执行查询并转换为列表
        hierarchy_data = list(result_query)
        print(hierarchy_data)

补充说明 #

• 数据库适配：若使用MySQL，原生SQL中的STRING_AGG需替换为GROUP_CONCAT，ORM方式中Django会自动处理适配。
• 性能优化：如果团队层级极深，可添加递归深度限制（比如在CTE中加WHERE lvl < 10），或缓存查询结果避免重复计算。

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