chore: merge feat/queue-based-graph-engine (#25833)

api/services/tools/tools_transform_service.py

@@ -1,6 +1,7 @@
 import json
 import logging
-from typing import Any, Union, cast
+from collections.abc import Mapping
+from typing import Any, Union
 
 from yarl import URL
 
@@ -39,7 +40,9 @@ class ToolTransformService:
         return str(url_prefix % {"tenant_id": tenant_id, "filename": filename})
 
     @classmethod
-    def get_tool_provider_icon_url(cls, provider_type: str, provider_name: str, icon: str | dict) -> Union[str, dict]:
+    def get_tool_provider_icon_url(
+        cls, provider_type: str, provider_name: str, icon: str | Mapping[str, str]
+    ) -> str | Mapping[str, str]:
         """
         get tool provider icon url
         """
@@ -52,7 +55,7 @@ class ToolTransformService:
         elif provider_type in {ToolProviderType.API.value, ToolProviderType.WORKFLOW.value}:
             try:
                 if isinstance(icon, str):
-                    return cast(dict, json.loads(icon))
+                    return json.loads(icon)
                 return icon
             except Exception:
                 return {"background": "#252525", "content": "\ud83d\ude01"}
@@ -119,7 +122,7 @@ class ToolTransformService:
             name=provider_controller.entity.identity.name,
             description=provider_controller.entity.identity.description,
             icon=provider_controller.entity.identity.icon,
-            icon_dark=provider_controller.entity.identity.icon_dark,
+            icon_dark=provider_controller.entity.identity.icon_dark or "",
             label=provider_controller.entity.identity.label,
             type=ToolProviderType.BUILT_IN,
             masked_credentials={},
@@ -141,9 +144,10 @@ class ToolTransformService:
             )
         }
 
+        masked_creds = {}
         for name in schema:
-            if result.masked_credentials:
-                result.masked_credentials[name] = ""
+            masked_creds[name] = ""
+        result.masked_credentials = masked_creds
 
         # check if the provider need credentials
         if not provider_controller.need_credentials:
@@ -221,7 +225,7 @@ class ToolTransformService:
             name=provider_controller.entity.identity.name,
             description=provider_controller.entity.identity.description,
             icon=provider_controller.entity.identity.icon,
-            icon_dark=provider_controller.entity.identity.icon_dark,
+            icon_dark=provider_controller.entity.identity.icon_dark or "",
             label=provider_controller.entity.identity.label,
             type=ToolProviderType.WORKFLOW,
             masked_credentials={},
@@ -334,7 +338,7 @@ class ToolTransformService:
 
     @staticmethod
     def convert_tool_entity_to_api_entity(
-        tool: Union[ApiToolBundle, WorkflowTool, Tool],
+        tool: ApiToolBundle | WorkflowTool | Tool,
         tenant_id: str,
         labels: list[str] | None = None,
     ) -> ToolApiEntity:
@@ -388,7 +392,7 @@ class ToolTransformService:
                 parameters=merged_parameters,
                 labels=labels or [],
             )
-        elif isinstance(tool, ApiToolBundle):
+        else:
             return ToolApiEntity(
                 author=tool.author,
                 name=tool.operation_id or "",
@@ -397,9 +401,6 @@ class ToolTransformService:
                 parameters=tool.parameters,
                 labels=labels or [],
             )
-        else:
-            # Handle WorkflowTool case
-            raise ValueError(f"Unsupported tool type: {type(tool)}")
 
     @staticmethod
     def convert_builtin_provider_to_credential_entity(

api/services/workflow_service.py

@@ -563,12 +563,12 @@ class WorkflowService:
             # This will prevent validation errors from breaking the workflow
             return []
 
-    def get_default_block_configs(self) -> list[dict]:
+    def get_default_block_configs(self) -> Sequence[Mapping[str, object]]:
         """
         Get default block configs
         """
         # return default block config
-        default_block_configs = []
+        default_block_configs: list[Mapping[str, object]] = []
         for node_class_mapping in NODE_TYPE_CLASSES_MAPPING.values():
             node_class = node_class_mapping[LATEST_VERSION]
             default_config = node_class.get_default_config()
@@ -577,7 +577,9 @@ class WorkflowService:
 
         return default_block_configs
 
-    def get_default_block_config(self, node_type: str, filters: dict | None = None) -> dict | None:
+    def get_default_block_config(
+        self, node_type: str, filters: Mapping[str, object] | None = None
+    ) -> Mapping[str, object]:
         """
         Get default config of node.
         :param node_type: node type
@@ -588,12 +590,12 @@ class WorkflowService:
 
         # return default block config
         if node_type_enum not in NODE_TYPE_CLASSES_MAPPING:
-            return None
+            return {}
 
         node_class = NODE_TYPE_CLASSES_MAPPING[node_type_enum][LATEST_VERSION]
         default_config = node_class.get_default_config(filters=filters)
         if not default_config:
-            return None
+            return {}
 
         return default_config
 
@@ -807,11 +809,13 @@ class WorkflowService:
                 WorkflowNodeExecutionStatus.EXCEPTION,
             )
             error = node_run_result.error if not run_succeeded else None
-
             return node, node_run_result, run_succeeded, error
-
         except WorkflowNodeRunFailedError as e:
-            return e._node, None, False, e._error
+            node = e.node
+            run_succeeded = False
+            node_run_result = None
+            error = e.error
+            return node, node_run_result, run_succeeded, error
 
     def _apply_error_strategy(self, node: Node, node_run_result: NodeRunResult) -> NodeRunResult:
         """Apply error strategy when node execution fails."""