在Django信号中运行CTE原生SQL查询报错求助

错误原因 #

你的SQL语句仅定义了hierarchy和lvls两个CTE（公共表表达式），但缺少主查询语句。CTE只是临时数据集的定义，必须搭配实际的查询逻辑来引用这些临时表，否则数据库会报语法错误（对应截图中的psycopg2.errors.SyntaxError: syntax error at end of input）。

修复步骤 #

1. 添加主查询
   在CTE定义后追加主查询，比如要获取lvls中的数据，就在SQL末尾添加查询语句：

   SELECT * FROM lvls;
   

   完整SQL示例：

   WITH hierarchy
   AS (
       SELECT step
           ,step_dependency
           ,1 AS lvl
       FROM Projects_workflowstepdependency
       WHERE step = %s
   ),
   lvls
   AS (
       SELECT lvl
           ,step
       FROM hierarchy
       GROUP BY lvl
           ,step
   )
   SELECT * FROM lvls; -- 新增主查询
   

2. 修正参数传递
   instance.workflow_step是WorkflowStep对象，SQL中step字段存储的是外键的UUID值，需要传递对象的id属性：

   [instance.workflow_step.id]
   

3. 避免硬编码表名（可选但推荐）
   直接用模型的_meta.db_table获取表名，避免因应用名/模型名变更导致表名错误：

   from .models import WorkflowStepDependency
   
   table_name = WorkflowStepDependency._meta.db_table
   

   对应SQL中替换硬编码表名：

   SELECT step, step_dependency, 1 AS lvl FROM {} WHERE step = %s
   

修复后的完整代码示例 #

@receiver(post_save, sender=WorkflowStepInstance)
def workflowStepInstanceUpdated(sender, instance, created, **kwargs):
    if not created:
        from .models import WorkflowStepDependency
        table_name = WorkflowStepDependency._meta.db_table
        
        with connection.cursor() as cursor:
            cursor.execute(f"""
                   WITH hierarchy
                   AS (
                       SELECT step
                           ,step_dependency
                           ,1 AS lvl
                       FROM {table_name}
                       WHERE step = %s

                       ),
                       lvls
                   AS (
                       SELECT lvl
                           ,step
                       FROM hierarchy
                       GROUP BY lvl
                           ,step
                       )
                   SELECT * FROM lvls; -- 主查询
                   """,
                   [instance.workflow_step.id]
                   )
            # 如需获取结果，可使用cursor.fetchall()
            # results = cursor.fetchall()

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