added tuple sequence edge input typing

Author: mathisxy

src/edgygraph/graphs.py

@@ -3,16 +3,17 @@
 from collections import defaultdict
 from collections.abc import Hashable
 import asyncio
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, SkipValidation
 
-from .nodes import START, Node
+from .nodes import START, END, Node # type: ignore
 from .states import StateProtocol as State, SharedProtocol as Shared
 from .hooks import GraphHook
 from .diff import Change, ChangeConflictException, Diff
 from .types import  \
     Edge, ErrorEdge, \
     SingleSource, SingleErrorSource, \
     Config, ErrorConfig, \
+    NodeTupel, is_node_tupel, \
     Entry, ErrorEntry, Entries, \
     NextNode
 
@@ -64,7 +65,7 @@ class Graph[T: State = State, S: Shared = Shared](BaseModel):
     - `(source, target, Config(instant=True))`: An instant edge from source to target. The target nodes are collected recursively and executed parallel to the source node. Make sure not to create cycles.
     - `(ValueError, target)`: An error edge from ValueError to target. The edge is traversed if a node, which is executed by an incoming edge located BEFORE this error edge in the edge list, throws a ValueError.
     - `((source, Exception), target)`: An error edge from Exception to target. The edge is traversed if the source node is executed by an incoming edge which is located BEFORE this error edge in the edge list throws an Exception. Source node lists are also supported.
-    - `(Exception, target, ErrorConfig(propagate=True))`: If propagate is True, the exception is propagated to the next error edges in the edge list. If the exception is not handled by any error edge, it is ultimately raised.
+    - `(Exception, target, ErrorConfig(propagate=True))`: If propagate is `True`, the exception is propagated to the next error edges in the edge list. If the exception is not handled by any error edge, it is ultimately raised.
 
 
     Attributes:
@@ -74,7 +75,7 @@ class Graph[T: State = State, S: Shared = Shared](BaseModel):
 
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
-    edges: Sequence[Edge[T, S] | ErrorEdge[T, S]] = Field(default_factory=list[Edge[T, S] | ErrorEdge[T, S]])
+    edges: Sequence[Edge[T, S] | ErrorEdge[T, S] | SkipValidation[NodeTupel[T, S]]] = Field(default_factory=list[Edge[T, S] | ErrorEdge[T, S] | NodeTupel[T, S]])
     hooks: Sequence[GraphHook[T, S]] = Field(default_factory=list[GraphHook[T, S]], exclude=True)
 
     edge_index: dict[SingleSource[T, S], list[Entry[T, S]]] = Field(default_factory=lambda: defaultdict(list), init=False)
@@ -101,6 +102,14 @@ def index_edges(self) -> None:
 
         for i, edge in enumerate(self.edges):
 
+            # Node Sequence
+            if is_node_tupel(edge):
+                edge = cast(NodeTupel[T, S], edge)
+                for source, next in zip(edge, edge[1:]):
+                    if isinstance(source, type): assert source is START, f"Unexpected type in node sequence: {source}"
+                    self.edge_index[source].append(Entry[T, S](next=next, config=Config(), index=i))
+                continue
+
             match edge:
                 case (source, next, config): pass
                 case (source, next): config = Config() if source is START or isinstance(source, (Node, list)) else ErrorConfig()
@@ -262,6 +271,7 @@ async def get_next_from_error(self, state: T, shared: S, eg: ExceptionGroup) ->
         for e in eg.exceptions:
 
             print(e)
+            
 
             source_node: NextNode[T, S] | None = getattr(e, "source_node", None)
 

src/edgygraph/types.py

@@ -6,6 +6,8 @@
 from .states import StateProtocol as State, SharedProtocol as Shared
 
 
+type NodeTupel[T: State, S: Shared] = tuple[type[START] | Node[T, S], *tuple[Node[T, S], ...]] | tuple[Node[T, S] | type[START], *tuple[Node[T, S], ...], type[END]]
+
 type SingleSource[T: State, S: Shared] = Node[T, S] | type[START]
 type Source[T: State, S: Shared] = SingleSource[T, S] | list[SingleSource[T, S]]
 
@@ -20,6 +22,29 @@
 type ErrorEdge[T: State, S: Shared] = tuple[ErrorSource[T, S], Next[T, S]] | tuple[ErrorSource[T, S], Next[T, S], ErrorConfig]
 
 
+def is_node_tupel(edge: tuple[Any, ...]) -> bool:
+
+    if len(edge) < 2:
+        return False
+
+    if is_only_node_tuple(edge):
+        return True
+    
+    if edge[0] is START and is_only_node_tuple(edge[1:]):
+        return True
+    
+    if edge[0] is START and is_only_node_tuple(edge[1:-1]) and edge[-1] is END:
+        return True
+    
+    if is_only_node_tuple(edge[:-1]) and edge[-1] is END:
+        return True
+    
+    return False
+
+def is_only_node_tuple(edge: tuple[Any, ...]) -> bool:
+    return all(isinstance(n, Node) for n in edge)
+
+
 class Config(BaseModel):
     """
     Configuration for the edge.