From 37bece696c47393d52a6b0ca770413190e30bd98 Mon Sep 17 00:00:00 2001 From: Asuka Minato Date: Sun, 20 Sep 2026 03:42:47 +0000 Subject: [PATCH] chore: use pydantic to verify (#38365) Co-authored-by: Byron.wang --- api/core/app/apps/workflow_app_runner.py | 147 ++++++++++++------ .../app/apps/test_workflow_app_runner_core.py | 80 ++++++++-- .../test_workflow_app_runner_single_node.py | 30 +--- 3 files changed, 172 insertions(+), 85 deletions(-) diff --git a/api/core/app/apps/workflow_app_runner.py b/api/core/app/apps/workflow_app_runner.py index 6399ea3d089..e4759d28ec2 100644 --- a/api/core/app/apps/workflow_app_runner.py +++ b/api/core/app/apps/workflow_app_runner.py @@ -3,7 +3,7 @@ import time from collections.abc import Mapping, Sequence from typing import Any, cast -from pydantic import ValidationError +from pydantic import BaseModel, ConfigDict, ValidationError from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom from core.app.entities.agent_strategy import AgentStrategyInfo @@ -56,7 +56,8 @@ from core.workflow.system_variables import ( from core.workflow.variable_pool_initializer import add_variables_to_pool from core.workflow.workflow_entry import WorkflowEntry from core.workflow.workflow_run_outputs import project_node_outputs_for_workflow_run -from graphon.entities.graph_config import NodeConfigDictAdapter +from graphon.entities.base_node_data import BaseNodeData +from graphon.entities.graph_config import NodeConfigDict from graphon.entities.pause_reason import HitlRequired from graphon.enums import BuiltinNodeTypes from graphon.graph import Graph @@ -96,6 +97,67 @@ from tasks.mail_human_input_delivery_task import dispatch_human_input_email_task logger = logging.getLogger(__name__) +class _WorkflowGraphNodeData(BaseNodeData): + """Node data fields the runner needs before concrete node validation.""" + + start_node_id: str | None = None + iteration_id: str | None = None + loop_id: str | None = None + + def parent_id_for(self, key: str) -> str | None: + match key: + case "iteration_id": + return self.iteration_id + case "loop_id": + return self.loop_id + case _: + return None + + +class _WorkflowGraphNodeConfig(BaseModel): + """Top-level node wrapper used for runner graph filtering.""" + + id: str + data: _WorkflowGraphNodeData + + model_config = ConfigDict(extra="allow", strict=True) + + def to_graph_node_config(self) -> dict[str, Any]: + return { + **self.model_dump(mode="python", exclude={"data"}, exclude_unset=True), + "data": self.data.model_dump(mode="python", exclude_unset=True), + } + + def to_typed_node_config(self) -> NodeConfigDict: + return cast(NodeConfigDict, {**self.model_dump(mode="python", exclude={"data"}), "data": self.data}) + + +class _WorkflowGraphEdgeConfig(BaseModel): + """Top-level edge wrapper used for selecting a single-node subgraph.""" + + source: str | None = None + target: str | None = None + + model_config = ConfigDict(extra="allow", strict=True) + + +class _WorkflowGraphConfig(BaseModel): + """Validated graph config boundary for runner methods. + + Workflow graph payloads are persisted JSON mappings with extra metadata used + by graphon, so validation is intentionally limited to the top-level graph + shape this runner needs before passing the mapping onward. + """ + + nodes: list[_WorkflowGraphNodeConfig] + edges: list[_WorkflowGraphEdgeConfig] + + model_config = ConfigDict(extra="allow", strict=True) + + def to_graph_config(self) -> dict[str, Any]: + return self.model_dump(mode="python", exclude_unset=True) + + class WorkflowBasedAppRunner: def __init__( self, @@ -132,14 +194,7 @@ class WorkflowBasedAppRunner: """ Init graph """ - if "nodes" not in graph_config or "edges" not in graph_config: - raise ValueError("nodes or edges not found in workflow graph") - - if not isinstance(graph_config.get("nodes"), list): - raise ValueError("nodes in workflow graph must be a list") - - if not isinstance(graph_config.get("edges"), list): - raise ValueError("edges in workflow graph must be a list") + graph_config = _WorkflowGraphConfig.model_validate(graph_config).to_graph_config() # Create explicit graph init context for Graph.init. run_context = build_dify_run_context( @@ -259,6 +314,8 @@ class WorkflowBasedAppRunner: ) -> tuple[Graph, VariablePool]: """ Get graph and variable pool for single node execution (iteration or loop). + The workflow graph and user inputs are treated as caller-owned data; the + narrowed graph config used for execution is built locally. Args: workflow: The workflow instance @@ -272,47 +329,39 @@ class WorkflowBasedAppRunner: A tuple containing (graph, variable_pool) """ # fetch workflow graph - graph_config = workflow.graph_dict - if not graph_config: - raise ValueError("workflow graph not found") + source_graph_config = _WorkflowGraphConfig.model_validate(workflow.graph_dict) - graph_config = cast(dict[str, Any], graph_config) - - if "nodes" not in graph_config or "edges" not in graph_config: - raise ValueError("nodes or edges not found in workflow graph") - - if not isinstance(graph_config.get("nodes"), list): - raise ValueError("nodes in workflow graph must be a list") - - if not isinstance(graph_config.get("edges"), list): - raise ValueError("edges in workflow graph must be a list") + node_user_inputs = dict(user_inputs) # filter nodes only in the specified node type (iteration or loop) - main_node_config = next((n for n in graph_config.get("nodes", []) if n.get("id") == node_id), None) - start_node_id = main_node_config.get("data", {}).get("start_node_id") if main_node_config else None - node_configs = [ + main_node_config = next((node for node in source_graph_config.nodes if node.id == node_id), None) + start_node_id = main_node_config.data.start_node_id if main_node_config else None + selected_node_configs = [ node - for node in graph_config.get("nodes", []) - if node.get("id") == node_id - or node.get("data", {}).get(node_type_filter_key, "") == node_id - or (start_node_id and node.get("id") == start_node_id) + for node in source_graph_config.nodes + if node.id == node_id + or node.data.parent_id_for(node_type_filter_key) == node_id + or (start_node_id and node.id == start_node_id) ] - graph_config["nodes"] = node_configs - - node_ids = [node.get("id") for node in node_configs] + node_ids = [node.id for node in selected_node_configs] # filter edges only in the specified node type - edge_configs = [ + selected_edge_configs = [ edge - for edge in graph_config.get("edges", []) - if (edge.get("source") is None or edge.get("source") in node_ids) - and (edge.get("target") is None or edge.get("target") in node_ids) + for edge in source_graph_config.edges + if (edge.source is None or edge.source in node_ids) and (edge.target is None or edge.target in node_ids) ] - graph_config["edges"] = edge_configs + node_configs = [node.to_graph_node_config() for node in selected_node_configs] + edge_configs = [edge.model_dump(mode="python", exclude_unset=True) for edge in selected_edge_configs] + graph_config = { + **source_graph_config.to_graph_config(), + "nodes": node_configs, + "edges": edge_configs, + } - typed_node_configs = [NodeConfigDictAdapter.validate_python(node) for node in node_configs] + typed_node_configs = selected_node_configs # Create explicit graph init context for Graph.init. run_context = build_dify_run_context( @@ -338,7 +387,7 @@ class WorkflowBasedAppRunner: target_node_config = None for node in typed_node_configs: - if node["id"] == node_id: + if node.id == node_id: target_node_config = node break @@ -346,8 +395,8 @@ class WorkflowBasedAppRunner: raise ValueError(f"{node_type_label} node id not found in workflow graph") # Get node class - node_type = target_node_config["data"].type - node_version = str(target_node_config["data"].version) + node_type = target_node_config.data.type + node_version = str(target_node_config.data.version) node_cls = resolve_workflow_node_class(node_type=node_type, node_version=node_version) # Use the variable pool from graph_runtime_state instead of creating a new one @@ -360,22 +409,22 @@ class WorkflowBasedAppRunner: selector for node_config in typed_node_configs for selector in get_node_creation_preload_selectors( - node_type=node_config["data"].type, - node_data=node_config["data"], + node_type=node_config.data.type, + node_data=node_config.data, ) ], ) try: variable_mapping = node_cls.extract_variable_selector_to_variable_mapping( - graph_config=workflow.graph_dict, config=target_node_config + graph_config=graph_config, config=target_node_config.to_typed_node_config() ) except NotImplementedError: variable_mapping = {} variable_mapping = inject_default_system_variable_mappings( - node_id=target_node_config["id"], + node_id=target_node_config.id, node_type=node_type, - node_data=target_node_config["data"], + node_data=target_node_config.data, variable_mapping=variable_mapping, ) @@ -383,12 +432,12 @@ class WorkflowBasedAppRunner: variable_loader=self._variable_loader, variable_pool=variable_pool, variable_mapping=variable_mapping, - user_inputs=user_inputs, + user_inputs=node_user_inputs, ) WorkflowEntry.mapping_user_inputs_to_variable_pool( variable_mapping=variable_mapping, - user_inputs=user_inputs, + user_inputs=node_user_inputs, variable_pool=variable_pool, tenant_id=workflow.tenant_id, ) diff --git a/api/tests/unit_tests/core/app/apps/test_workflow_app_runner_core.py b/api/tests/unit_tests/core/app/apps/test_workflow_app_runner_core.py index 94dac233314..c2ecd33d323 100644 --- a/api/tests/unit_tests/core/app/apps/test_workflow_app_runner_core.py +++ b/api/tests/unit_tests/core/app/apps/test_workflow_app_runner_core.py @@ -1,10 +1,12 @@ from __future__ import annotations import json +from copy import deepcopy from datetime import UTC, datetime from types import SimpleNamespace import pytest +from pydantic import ValidationError from core.app.apps.workflow_app_runner import WorkflowBasedAppRunner from core.app.entities.app_invoke_entities import DIFY_RUN_CONTEXT_KEY, InvokeFrom, UserFrom @@ -78,7 +80,7 @@ class TestWorkflowBasedAppRunner: start_at=0.0, ) - with pytest.raises(ValueError, match="nodes or edges not found"): + with pytest.raises(ValidationError, match="nodes"): runner._init_graph( graph_config={}, graph_runtime_state=runtime_state, @@ -86,7 +88,7 @@ class TestWorkflowBasedAppRunner: invoke_from=InvokeFrom.DEBUGGER, ) - with pytest.raises(ValueError, match="nodes in workflow graph must be a list"): + with pytest.raises(ValidationError, match="nodes"): runner._init_graph( graph_config={"nodes": {}, "edges": []}, graph_runtime_state=runtime_state, @@ -94,7 +96,7 @@ class TestWorkflowBasedAppRunner: invoke_from=InvokeFrom.DEBUGGER, ) - with pytest.raises(ValueError, match="edges in workflow graph must be a list"): + with pytest.raises(ValidationError, match="edges"): runner._init_graph( graph_config={"nodes": [], "edges": {}}, graph_runtime_state=runtime_state, @@ -192,6 +194,71 @@ class TestWorkflowBasedAppRunner: assert graph is not None assert variable_pool is graph_runtime_state.variable_pool + def test_get_graph_and_variable_pool_for_single_node_run_does_not_mutate_inputs( + self, monkeypatch: pytest.MonkeyPatch + ): + runner = WorkflowBasedAppRunner(queue_manager=SimpleNamespace(), app_id="app") + graph_runtime_state = GraphRuntimeState( + variable_pool=VariablePool.from_bootstrap(system_variables=default_system_variables()), + start_at=0.0, + ) + graph_config = { + "nodes": [ + {"id": "node-1", "data": {"type": "start", "version": "1"}}, + {"id": "node-child", "data": {"type": "start", "version": "1", "iteration_id": "node-1"}}, + {"id": "outside", "data": {"type": "start", "version": "1"}}, + ], + "edges": [ + {"id": "edge-1", "source": "node-1", "target": "node-child"}, + {"id": "edge-2", "source": "outside", "target": "node-1"}, + ], + } + expected_graph_config = deepcopy(graph_config) + workflow = SimpleNamespace(tenant_id="tenant", id="workflow", graph_dict=graph_config) + user_inputs = {"question": "hello"} + expected_user_inputs = dict(user_inputs) + captured_graph_config = {} + + def fake_graph_init(**kwargs): + captured_graph_config["value"] = kwargs["graph_config"] + return SimpleNamespace() + + class _NodeCls: + @staticmethod + def extract_variable_selector_to_variable_mapping(graph_config, config): + return {} + + from core.app.apps import workflow_app_runner + + monkeypatch.setattr("core.app.apps.workflow_app_runner.Graph.init", fake_graph_init) + monkeypatch.setattr(workflow_app_runner, "resolve_workflow_node_class", lambda **_kwargs: _NodeCls) + monkeypatch.setattr( + "core.app.apps.workflow_app_runner.load_into_variable_pool", + lambda **kwargs: kwargs["user_inputs"].update({"from_loader": "local-only"}), + ) + monkeypatch.setattr( + "core.app.apps.workflow_app_runner.WorkflowEntry.mapping_user_inputs_to_variable_pool", + lambda **kwargs: kwargs["user_inputs"].update({"from_mapping": "local-only"}), + ) + + graph, variable_pool = runner._get_graph_and_variable_pool_for_single_node_run( + workflow=workflow, + node_id="node-1", + user_inputs=user_inputs, + graph_runtime_state=graph_runtime_state, + node_type_filter_key="iteration_id", + node_type_label="iteration", + user_id="00000000-0000-0000-0000-000000000001", + ) + + assert graph is not None + assert variable_pool is graph_runtime_state.variable_pool + assert workflow.graph_dict == expected_graph_config + assert user_inputs == expected_user_inputs + assert captured_graph_config["value"] is not workflow.graph_dict + assert [node["id"] for node in captured_graph_config["value"]["nodes"]] == ["node-1", "node-child"] + assert captured_graph_config["value"]["edges"] == [{"id": "edge-1", "source": "node-1", "target": "node-child"}] + def test_get_graph_and_variable_pool_for_single_node_run_includes_trace_session_id( self, monkeypatch: pytest.MonkeyPatch ): @@ -294,18 +361,11 @@ class TestWorkflowBasedAppRunner: def extract_variable_selector_to_variable_mapping(graph_config, config): return {} - def _validate_node_config(value): - return {"id": value["id"], "data": SimpleNamespace(**value["data"])} - def _graph_init(**kwargs): variable_pool = graph_runtime_state.variable_pool assert variable_pool.get(["sys", "conversation_id"]) is not None return SimpleNamespace() - monkeypatch.setattr( - "core.app.apps.workflow_app_runner.NodeConfigDictAdapter.validate_python", - _validate_node_config, - ) monkeypatch.setattr( "core.app.apps.workflow_app_runner.Graph.init", _graph_init, diff --git a/api/tests/unit_tests/core/app/apps/test_workflow_app_runner_single_node.py b/api/tests/unit_tests/core/app/apps/test_workflow_app_runner_single_node.py index 3ae4462f9b8..baa831ee689 100644 --- a/api/tests/unit_tests/core/app/apps/test_workflow_app_runner_single_node.py +++ b/api/tests/unit_tests/core/app/apps/test_workflow_app_runner_single_node.py @@ -5,6 +5,7 @@ from typing import Any from unittest.mock import MagicMock, patch import pytest +from pydantic import ValidationError from core.app.apps.base_app_queue_manager import AppQueueManager from core.app.apps.workflow.app_runner import WorkflowAppRunner @@ -12,7 +13,6 @@ from core.app.apps.workflow_app_runner import WorkflowBasedAppRunner from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity from core.credit_usage import CreditUsageAppType from core.workflow.system_variables import default_system_variables -from graphon.entities.graph_config import NodeConfigDictAdapter from graphon.runtime import GraphRuntimeState, VariablePool from models.model import AppMode from models.workflow import Workflow, WorkflowKind @@ -118,7 +118,7 @@ def test_run_uses_single_node_execution_branch( assert entry_kwargs["graph_runtime_state"] is graph_runtime_state -def test_single_node_run_validates_target_node_config(monkeypatch: pytest.MonkeyPatch) -> None: +def test_single_node_run_validates_target_node_config() -> None: runner = WorkflowBasedAppRunner( queue_manager=MagicMock(spec=AppQueueManager), variable_loader=MagicMock(), @@ -133,14 +133,7 @@ def test_single_node_run_validates_target_node_config(monkeypatch: pytest.Monkey "nodes": [ { "id": "loop-node", - "data": { - "type": "loop", - "title": "Loop", - "loop_count": 1, - "start_node_id": "loop-start", - "break_conditions": [], - "logical_operator": "and", - }, + "data": {}, } ], "edges": [], @@ -149,21 +142,8 @@ def test_single_node_run_validates_target_node_config(monkeypatch: pytest.Monkey ) _, _, graph_runtime_state = _make_graph_state() - seen_configs: list[object] = [] - original_validate_python = NodeConfigDictAdapter.validate_python - def record_validate_python(value: object): - seen_configs.append(value) - return original_validate_python(value) - - monkeypatch.setattr(NodeConfigDictAdapter, "validate_python", record_validate_python) - - with ( - patch("core.app.apps.workflow_app_runner.DifyNodeFactory"), - patch("core.app.apps.workflow_app_runner.Graph.init", return_value=MagicMock()), - patch("core.app.apps.workflow_app_runner.load_into_variable_pool"), - patch("core.app.apps.workflow_app_runner.WorkflowEntry.mapping_user_inputs_to_variable_pool"), - ): + with pytest.raises(ValidationError, match="nodes.0.data.type"): runner._get_graph_and_variable_pool_for_single_node_run( workflow=workflow, node_id="loop-node", @@ -174,8 +154,6 @@ def test_single_node_run_validates_target_node_config(monkeypatch: pytest.Monkey user_id="00000000-0000-0000-0000-000000000001", ) - assert seen_configs == [workflow.graph_dict["nodes"][0]] - def test_run_adds_inputs_with_snippet_compatible_start_aliases() -> None: app_config = MagicMock()