From f5ce1ec941076d24f7b51f24f3e0a4f92bff58bf Mon Sep 17 00:00:00 2001 From: Lars Date: Mon, 13 Apr 2026 08:45:55 +0200 Subject: [PATCH] refactor: Proper type-safe condition handling with Union types MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Previous fix used Any type, breaking type safety and only handling simple cases. This is the correct implementation: Changes: 1. LogicExpression.operands: List[Any] → List['LogicExpression'] - Enables recursive/nested expressions - Proper type checking for all operator combinations 2. WorkflowNode.condition: Any → Union[LogicExpression, Condition] - Type-safe deserialization - Supports both UI format (direct LogicExpression) and legacy (Condition wrapper) - Pydantic automatically tries LogicExpression first, then Condition 3. Executor: Simplified with isinstance() checks - Clean type detection without dict manipulation - Fallback for edge cases This now correctly handles: - Simple conditions: {operator: "eq", ref: "...", value: "..."} - Combined: {operator: "and", operands: [...]} - Nested: {operator: "or", operands: [{operator: "and", ...}, ...]} - All operators: eq, neq, in, not_in, gt, lt, gte, lte, contains, and, or, not - Legacy format: {expression: {...}, then_path: "...", else_path: "..."} Co-Authored-By: Claude Opus 4.6 --- backend/workflow_executor.py | 37 ++++++++++++++---------------------- backend/workflow_models.py | 9 ++++++--- 2 files changed, 20 insertions(+), 26 deletions(-) diff --git a/backend/workflow_executor.py b/backend/workflow_executor.py index 1aa67d5..dbc398b 100644 --- a/backend/workflow_executor.py +++ b/backend/workflow_executor.py @@ -409,37 +409,28 @@ def execute_logic_node( completed_at=datetime.utcnow().isoformat() ) - # Handle both serialization formats: - # UI format: condition = {operands: [...], operator: "and"} (dict or LogicExpression) - # Legacy format: condition = {expression: {operands: [...], operator: "and"}} (Condition object) + # Handle both formats (thanks to Union[LogicExpression, Condition] type): + # 1. Direct LogicExpression (UI format): node.condition is LogicExpression + # 2. Wrapped in Condition (legacy): node.condition is Condition with .expression + from workflow_models import LogicExpression, Condition + expression = None - # Convert to dict if it's a Pydantic model - condition_dict = node.condition - if hasattr(node.condition, 'model_dump'): - condition_dict = node.condition.model_dump() - elif hasattr(node.condition, 'dict'): - condition_dict = node.condition.dict() - - # Check if it's a dict - if isinstance(condition_dict, dict): + if isinstance(node.condition, LogicExpression): # UI format: direct LogicExpression - if 'operator' in condition_dict and 'operands' in condition_dict: - from workflow_models import LogicExpression - expression = LogicExpression(**condition_dict) + expression = node.condition + elif isinstance(node.condition, Condition): # Legacy format: wrapped in Condition - elif 'expression' in condition_dict and condition_dict['expression'] is not None: - from workflow_models import LogicExpression - expression = LogicExpression(**condition_dict['expression']) - # Pydantic object + expression = node.condition.expression else: + # Fallback: try to detect format manually if hasattr(node.condition, 'operator') and hasattr(node.condition, 'operands'): - expression = node.condition - elif hasattr(node.condition, 'expression') and node.condition.expression is not None: - expression = node.condition.expression + expression = node.condition # Looks like LogicExpression + elif hasattr(node.condition, 'expression'): + expression = node.condition.expression # Looks like Condition if expression is None: - error_msg = f"Logic node {node.id} has invalid or empty condition (operator/operands/expression is None or missing)" + error_msg = f"Logic node {node.id} has no valid condition/expression defined" logger.error(error_msg) return NodeExecutionState( node_id=node.id, diff --git a/backend/workflow_models.py b/backend/workflow_models.py index e98c0fe..76ace63 100644 --- a/backend/workflow_models.py +++ b/backend/workflow_models.py @@ -6,7 +6,7 @@ Data validation schemas for Workflow-Graph, Knoten, Kanten, Bedingungen. Konzept-Basis: konzept_workflow_engine_konsolidated.md Anforderungsanalyse: anforderungsanalyse_umsetzungsplan.md """ -from typing import Optional, List, Dict, Any +from typing import Optional, List, Dict, Any, Union from pydantic import BaseModel, Field from enum import Enum @@ -148,11 +148,14 @@ class LogicExpression(BaseModel): } """ operator: LogicOperator = Field(..., description="Logischer Operator (and, or, not) oder Vergleichsoperator") - operands: Optional[List[Any]] = Field(None, description="Liste von Operanden (LogicOperand oder verschachtelte LogicExpression)") + operands: Optional[List['LogicExpression']] = Field(None, description="Liste von Operanden (LogicOperand oder verschachtelte LogicExpression)") # Bei einfachem Vergleich: ref: Optional[str] = Field(None, description="Signal-Referenz (nur bei Vergleichsoperatoren)") value: Optional[Any] = Field(None, description="Vergleichswert (nur bei Vergleichsoperatoren)") +# Enable forward reference resolution for recursive model +LogicExpression.model_rebuild() + class Condition(BaseModel): """ @@ -196,7 +199,7 @@ class WorkflowNode(BaseModel): # LOGIC-Knoten # Support both formats: direct LogicExpression (UI) or wrapped in Condition (legacy) - condition: Optional[Any] = Field(None, description="Bedingung für Pfad-Routing (LogicExpression or Condition)") + condition: Optional[Union[LogicExpression, Condition]] = Field(None, description="Bedingung für Pfad-Routing") fallback: Optional[FallbackConfig] = Field(None, description="Fallback-Konfiguration") # JOIN-Knoten (Phase 4)