refactor(workflow): simplify if-else node condition structure by removing nested condition groups

The changes remove the `ConditionGroup` abstraction and flatten condition expressions directly under `ConditionBranchConfig.expressions`. This simplifies the data model and evaluation logic, eliminating redundant grouping layers while preserving all functionality. The migration logic and group-level operators are removed as they are no longer needed.

BREAKING CHANGE: `ConditionBranchConfig.expressions` now expects a flat list of `ConditionDetail` instead of `ConditionGroup`; existing configurations must be updated to use direct condition lists.
This commit is contained in:
Timebomb2018
2026-04-17 17:46:49 +08:00
parent 62c721bdf6
commit fb93c509f4
2 changed files with 36 additions and 77 deletions

View File

@@ -1,6 +1,6 @@
"""Condition Configuration"""
from typing import Any
from pydantic import Field, BaseModel, field_validator, model_validator
from pydantic import Field, BaseModel, field_validator
from app.core.workflow.nodes.base_config import BaseNodeConfig
from app.core.workflow.nodes.enums import ComparisonOperator, LogicOperator, ValueInputType
@@ -57,49 +57,22 @@ class ConditionDetail(BaseModel):
return v
class ConditionGroup(BaseModel):
"""A group of conditions combined by group_operator (AND/OR)"""
group_operator: LogicOperator = Field(
default=LogicOperator.AND,
description="Logical operator used to combine conditions within this group"
)
conditions: list[ConditionDetail] = Field(
...,
description="List of conditions within this group"
)
class ConditionBranchConfig(BaseModel):
"""Configuration for a conditional branch.
logical_operator controls how groups are combined.
Each group's group_operator controls how conditions within it are combined.
logical_operator controls how all expressions are combined (AND/OR).
"""
logical_operator: LogicOperator = Field(
default=LogicOperator.AND,
description="Logical operator used to combine condition groups"
description="Logical operator used to combine all conditions"
)
expressions: list[ConditionGroup] = Field(
...,
description="List of condition groups within this branch"
expressions: list[ConditionDetail] = Field(
default_factory=list,
description="List of conditions within this branch"
)
@model_validator(mode="before")
@classmethod
def migrate_flat_expressions(cls, data):
"""Migrate legacy flat expressions (list[ConditionDetail]) to list[ConditionGroup]."""
exprs = data.get("expressions", [])
if exprs and isinstance(exprs[0], dict) and "left" in exprs[0]:
data["expressions"] = [{
"group_operator": data.get("logical_operator", "and"),
"conditions": exprs
}]
return data
class IfElseNodeConfig(BaseNodeConfig):
cases: list[ConditionBranchConfig] = Field(

View File

@@ -26,28 +26,20 @@ class IfElseNode(BaseNode):
def _extract_input(self, state: WorkflowState, variable_pool: VariablePool) -> dict[str, Any]:
result = []
for case in self.typed_config.cases:
groups = []
for group in case.expressions:
conditions = []
for condition in group.conditions:
conditions.append({
"left": self.get_variable(condition.left, variable_pool, strict=False),
"right": condition.right
if condition.input_type == ValueInputType.CONSTANT or condition.right is None
else self.get_variable(condition.right, variable_pool, strict=False),
"operator": str(condition.operator),
})
groups.append({
"group_operator": str(group.group_operator),
"conditions": conditions,
conditions = []
for condition in case.expressions:
conditions.append({
"left": self.get_variable(condition.left, variable_pool, strict=False),
"right": condition.right
if condition.input_type == ValueInputType.CONSTANT or condition.right is None
else self.get_variable(condition.right, variable_pool, strict=False),
"operator": str(condition.operator),
})
result.append({
"expressions": groups,
"expressions": conditions,
"logical_operator": str(case.logical_operator),
})
return {
"cases": result
}
return {"cases": result}
@staticmethod
def _evaluate(operator, instance: CompareOperatorInstance) -> Any:
@@ -99,36 +91,30 @@ class IfElseNode(BaseNode):
conditions = []
for case_branch in self.typed_config.cases:
group_results = []
for group in case_branch.expressions:
condition_results = []
for condition in group.conditions:
pattern = r"\{\{\s*(.*?)\s*\}\}"
left_string = re.sub(pattern, r"\1", condition.left).strip()
try:
left_value = self.get_variable(left_string, variable_pool)
except KeyError:
left_value = None
condition_results = []
for condition in case_branch.expressions:
pattern = r"\{\{\s*(.*?)\s*\}\}"
left_string = re.sub(pattern, r"\1", condition.left).strip()
try:
left_value = self.get_variable(left_string, variable_pool)
except KeyError:
left_value = None
# array[file] + sub_variable_condition: use ArrayFileContainsOperator directly
if condition.sub_variable_condition is not None and isinstance(left_value, list):
evaluator = ArrayFileContainsOperator(left_value, condition.sub_variable_condition)
else:
evaluator = ConditionExpressionResolver.resolve_by_value(left_value)(
variable_pool,
condition.left,
condition.right,
condition.input_type
)
condition_results.append(self._evaluate(condition.operator, evaluator))
if group.group_operator == LogicOperator.AND:
group_results.append(all(condition_results))
if condition.sub_variable_condition is not None and isinstance(left_value, list):
evaluator = ArrayFileContainsOperator(left_value, condition.sub_variable_condition)
else:
group_results.append(any(condition_results))
evaluator = ConditionExpressionResolver.resolve_by_value(left_value)(
variable_pool,
condition.left,
condition.right,
condition.input_type
)
condition_results.append(self._evaluate(condition.operator, evaluator))
if case_branch.logical_operator == LogicOperator.AND:
conditions.append(all(group_results))
conditions.append(all(condition_results))
else:
condition_res = any(group_results)
condition_res = any(condition_results)
conditions.append(condition_res)
if condition_res:
return conditions