From 7cbf4093f452955dac32bea5415872a5ebc0de15 Mon Sep 17 00:00:00 2001 From: -LAN- Date: Thu, 28 Aug 2025 04:30:50 +0800 Subject: [PATCH] chore(graph_engine): Use `TYPE | None` instead of `Optional` Signed-off-by: -LAN- --- api/core/workflow/graph_engine/domain/graph_execution.py | 3 +-- .../workflow/graph_engine/error_handling/abort_strategy.py | 3 +-- .../graph_engine/error_handling/default_value_strategy.py | 4 +--- .../graph_engine/error_handling/fail_branch_strategy.py | 4 +--- .../workflow/graph_engine/error_handling/retry_strategy.py | 3 +-- 5 files changed, 5 insertions(+), 12 deletions(-) diff --git a/api/core/workflow/graph_engine/domain/graph_execution.py b/api/core/workflow/graph_engine/domain/graph_execution.py index b8fa801289..c375b08fe0 100644 --- a/api/core/workflow/graph_engine/domain/graph_execution.py +++ b/api/core/workflow/graph_engine/domain/graph_execution.py @@ -3,7 +3,6 @@ GraphExecution aggregate root managing the overall graph execution state. """ from dataclasses import dataclass, field -from typing import Optional from .node_execution import NodeExecution @@ -21,7 +20,7 @@ class GraphExecution: started: bool = False completed: bool = False aborted: bool = False - error: Optional[Exception] = None + error: Exception | None = None node_executions: dict[str, NodeExecution] = field(default_factory=dict) def start(self) -> None: diff --git a/api/core/workflow/graph_engine/error_handling/abort_strategy.py b/api/core/workflow/graph_engine/error_handling/abort_strategy.py index e747704fda..ba2a7b514f 100644 --- a/api/core/workflow/graph_engine/error_handling/abort_strategy.py +++ b/api/core/workflow/graph_engine/error_handling/abort_strategy.py @@ -3,7 +3,6 @@ Abort error strategy implementation. """ import logging -from typing import Optional from core.workflow.graph import Graph from core.workflow.graph_events import GraphNodeEventBase, NodeRunFailedEvent @@ -19,7 +18,7 @@ class AbortStrategy: It stops the entire graph execution when a node fails. """ - def handle_error(self, event: NodeRunFailedEvent, graph: Graph, retry_count: int) -> Optional[GraphNodeEventBase]: + def handle_error(self, event: NodeRunFailedEvent, graph: Graph, retry_count: int) -> GraphNodeEventBase | None: """ Handle error by aborting execution. diff --git a/api/core/workflow/graph_engine/error_handling/default_value_strategy.py b/api/core/workflow/graph_engine/error_handling/default_value_strategy.py index 92e61dc22a..19e5b4c0ae 100644 --- a/api/core/workflow/graph_engine/error_handling/default_value_strategy.py +++ b/api/core/workflow/graph_engine/error_handling/default_value_strategy.py @@ -2,8 +2,6 @@ Default value error strategy implementation. """ -from typing import Optional - from core.workflow.enums import ErrorStrategy, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus from core.workflow.graph import Graph from core.workflow.graph_events import GraphNodeEventBase, NodeRunExceptionEvent, NodeRunFailedEvent @@ -18,7 +16,7 @@ class DefaultValueStrategy: predefined default output values. """ - def handle_error(self, event: NodeRunFailedEvent, graph: Graph, retry_count: int) -> Optional[GraphNodeEventBase]: + def handle_error(self, event: NodeRunFailedEvent, graph: Graph, retry_count: int) -> GraphNodeEventBase | None: """ Handle error by using default values. diff --git a/api/core/workflow/graph_engine/error_handling/fail_branch_strategy.py b/api/core/workflow/graph_engine/error_handling/fail_branch_strategy.py index 82e434c89b..d3be64539e 100644 --- a/api/core/workflow/graph_engine/error_handling/fail_branch_strategy.py +++ b/api/core/workflow/graph_engine/error_handling/fail_branch_strategy.py @@ -2,8 +2,6 @@ Fail branch error strategy implementation. """ -from typing import Optional - from core.workflow.enums import ErrorStrategy, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus from core.workflow.graph import Graph from core.workflow.graph_events import GraphNodeEventBase, NodeRunExceptionEvent, NodeRunFailedEvent @@ -18,7 +16,7 @@ class FailBranchStrategy: through a designated fail-branch edge. """ - def handle_error(self, event: NodeRunFailedEvent, graph: Graph, retry_count: int) -> Optional[GraphNodeEventBase]: + def handle_error(self, event: NodeRunFailedEvent, graph: Graph, retry_count: int) -> GraphNodeEventBase | None: """ Handle error by taking the fail branch. diff --git a/api/core/workflow/graph_engine/error_handling/retry_strategy.py b/api/core/workflow/graph_engine/error_handling/retry_strategy.py index 5956a7c62e..af1058569f 100644 --- a/api/core/workflow/graph_engine/error_handling/retry_strategy.py +++ b/api/core/workflow/graph_engine/error_handling/retry_strategy.py @@ -3,7 +3,6 @@ Retry error strategy implementation. """ import time -from typing import Optional from core.workflow.graph import Graph from core.workflow.graph_events import GraphNodeEventBase, NodeRunFailedEvent, NodeRunRetryEvent @@ -17,7 +16,7 @@ class RetryStrategy: maximum number of retries with configurable intervals. """ - def handle_error(self, event: NodeRunFailedEvent, graph: Graph, retry_count: int) -> Optional[GraphNodeEventBase]: + def handle_error(self, event: NodeRunFailedEvent, graph: Graph, retry_count: int) -> GraphNodeEventBase | None: """ Handle error by retrying the node.