minor: detail 接口支持subprocess_simple_inputs参数 (#7053)

* minor: detail 接口支持retry参数

* minor: detail 接口支持subprocess_simple_inputs参数控制是否简化独立子流程的返回

* test: fix 单元测试

gcloud/taskflow3/apis/django/api.py

@@ -200,14 +200,22 @@ def detail(request, project_id):
     task_id = request.query_params["instance_id"]
     node_id = request.query_params["node_id"]
     loop = request.query_params.get("loop")
+    subprocess_simple_inputs = request.query_params.get("subprocess_simple_inputs") == "true"
     component_code = request.query_params.get("component_code")
     include_data = int(request.query_params.get("include_data", 1))
 
     subprocess_stack = json.loads(request.query_params.get("subprocess_stack", "[]"))
 
     task = TaskFlowInstance.objects.get(pk=task_id, project_id=project_id)
     ctx = task.get_node_detail(
-        node_id, request.user.username, component_code, subprocess_stack, loop, include_data, project_id=project_id
+        node_id,
+        request.user.username,
+        component_code,
+        subprocess_stack,
+        loop,
+        include_data,
+        project_id=project_id,
+        subprocess_simple_inputs=subprocess_simple_inputs,
     )
 
     return JsonResponse(ctx)

gcloud/taskflow3/domains/dispatchers/node.py

@@ -504,7 +504,11 @@ def get_node_data_v2(
                 if node_info["type"] == "SubProcess":
                     # remove prefix '${' and subfix '}' in subprocess execution input
                     inputs = {k[2:-1]: v for k, v in data["inputs"].items()}
-                elif node_info["type"] == "ServiceActivity" and node_code == "subprocess_plugin":
+                elif (
+                    node_info["type"] == "ServiceActivity"
+                    and node_code == "subprocess_plugin"
+                    and kwargs.get("subprocess_simple_inputs")
+                ):
                     raw_inputs = data["inputs"]["subprocess"]["pipeline"]["constants"]
                     inputs = {key[2:-1]: value.get("value") for key, value in raw_inputs.items()}
                 else:
@@ -567,6 +571,7 @@ def get_node_data_v2(
                         subprocess_stack=subprocess_stack,
                         root_pipeline_data=root_pipeline_data,
                         parent_params=root_pipeline_context,
+                        subprocess_simple_inputs=kwargs.get("subprocess_simple_inputs"),
                     )
                 except Exception as e:
                     message = _(f"节点数据请求失败: 请重试, 如多次失败可联系管理员处理. {e} | get_node_data_v2")
@@ -581,7 +586,11 @@ def get_node_data_v2(
                 if node_info["type"] == "SubProcess":
                     # remove prefix '${' and subfix '}' in subprocess execution input
                     inputs = {k[2:-1]: v for k, v in preview_inputs.items()}
-                elif node_info["type"] == "ServiceActivity" and node_code == "subprocess_plugin":
+                elif (
+                    node_info["type"] == "ServiceActivity"
+                    and node_code == "subprocess_plugin"
+                    and kwargs.get("subprocess_simple_inputs")
+                ):
                     inputs = {k[2:-1]: v for k, v in preview_inputs.items()}
                 else:
                     inputs = preview_inputs
@@ -769,7 +778,11 @@ def get_node_detail_v2(
 
             for hist in detail["histories"]:
                 # 重试记录必然是因为失败才重试
-                if node_info["type"] == "ServiceActivity" and node_code == "subprocess_plugin":
+                if (
+                    node_info["type"] == "ServiceActivity"
+                    and node_code == "subprocess_plugin"
+                    and kwargs.get("subprocess_simple_inputs")
+                ):
                     # 对于独立子流程节点要重新渲染输出
                     raw_inputs = hist["inputs"]["subprocess"]["pipeline"]["constants"]
                     hist["inputs"] = {key[2:-1]: value.get("value") for key, value in raw_inputs.items()}

gcloud/taskflow3/models.py

@@ -958,6 +958,7 @@ def get_node_detail(
                 pipeline_instance=self.pipeline_instance,
                 subprocess_stack=subprocess_stack,
                 project_id=kwargs["project_id"],
+                subprocess_simple_inputs=kwargs.get("subprocess_simple_inputs", False),
             )
             if not node_data_result["result"]:
                 return node_data_result
@@ -969,6 +970,7 @@ def get_node_detail(
             loop=loop,
             pipeline_instance=self.pipeline_instance,
             subprocess_stack=subprocess_stack,
+            subprocess_simple_inputs=kwargs.get("subprocess_simple_inputs", False),
         )
         if not node_detail_result["result"]:
             return node_detail_result

gcloud/tasktmpl3/domains/constants.py

@@ -81,6 +81,7 @@ def preview_node_inputs(
     subprocess_stack: List[str] = [],
     root_pipeline_data: dict = {},
     parent_params: dict = {},
+    subprocess_simple_inputs: bool = False,
 ):
     def get_need_render_context_keys():
         keys = set()
@@ -130,11 +131,12 @@ def get_need_render_context_keys():
             subprocess_stack=subprocess_stack[1:],
             root_pipeline_data=root_pipeline_data,
             parent_params=formatted_param_data,
+            subprocess_simple_inputs=subprocess_simple_inputs,
         )
 
     if node_type == NodeType.ServiceActivity.value:
         # 如果是独立子流程
-        if node_code == "subprocess_plugin":
+        if node_code == "subprocess_plugin" and subprocess_simple_inputs:
             raw_inputs = pipeline["activities"][node_id]["component"]["inputs"]["subprocess"]["value"]["pipeline"][
                 "constants"
             ]

gcloud/tests/taskflow3/models/taskflow/test_get_node_detail.py

@@ -15,7 +15,6 @@
 
 from gcloud import err_code
 from gcloud.taskflow3.models import TaskFlowInstance
-
 from gcloud.tests.mock import *  # noqa
 from gcloud.tests.mock_settings import *  # noqa
 
@@ -97,6 +96,7 @@ def test_include_data_is_false(self):
             subprocess_stack=subprocess_stack,
             pipeline_instance=taskflow.pipeline_instance,
             loop=loop,
+            subprocess_simple_inputs=False,
         )
         self.assertEqual(detail, {"code": 0, "data": {}, "message": "", "result": True})
 
@@ -138,13 +138,15 @@ def test_success(self):
             pipeline_instance=taskflow.pipeline_instance,
             loop=loop,
             project_id="project_id",
+            subprocess_simple_inputs=False,
         )
         dispatcher.get_node_detail.assert_called_once_with(
             username=username,
             component_code=component_code,
             subprocess_stack=subprocess_stack,
             pipeline_instance=taskflow.pipeline_instance,
             loop=loop,
+            subprocess_simple_inputs=False,
         )
         self.assertEqual(
             detail, {"code": 0, "data": {"data": "data", "detail": "detail"}, "message": "", "result": True}