changed edge signature: removed END parameter because of redundancy with None

Author: mathisxy

src/edgygraph/graph/branches.py

@@ -4,10 +4,9 @@
 from collections.abc import Hashable, Sequence
 import asyncio
 
-from ..nodes import START, END, Node
 from ..states import StateProtocol, SharedProtocol
 from ..diff import Change
-from .types import NodeTupel, Edge, ErrorEdge, SingleNext, Entry, ErrorEntry, SingleSource, SingleErrorSource, Config, ErrorConfig, Types
+from .types import NodeTupel, Edge, ErrorEdge, Join, Entry, ErrorEntry, SingleSource, SingleErrorSource, Config, ErrorConfig, Types
 
 
 class Branch[T: StateProtocol, S: SharedProtocol]:
@@ -22,7 +21,7 @@ class Branch[T: StateProtocol, S: SharedProtocol]:
     """
 
 
-    def __init__(self, edges: Sequence[Edge[T, S] | ErrorEdge[T, S] | NodeTupel[T, S]], start: SingleSource[T, S], join: SingleNext[T, S] = None) -> None:
+    def __init__(self, edges: Sequence[Edge[T, S] | ErrorEdge[T, S] | NodeTupel[T, S]], start: SingleSource[T, S], join: Join[T, S] = None) -> None:
 
         self.edges = edges
         self.start = start
@@ -51,9 +50,8 @@ def index_edges(self) -> None:
             if Types[T, S].is_node_tupel(edge):
 
                 for source, next in zip(edge, edge[1:]):
-                    if isinstance(source, type): assert source is START, f"Unexpected type in node sequence: {source}"
-                    if isinstance(next, type): assert next is END, f"Unexpected type in node sequence: {next}"
-                    assert isinstance(source, (Node, type)), f"Unexpected source type in node sequence: {source}"
+                    assert Types[T, S].is_single_source(source), f"Unexpected source type in node sequence: {source}"
+                    assert Types[T, S].is_next(next), f"Unexpected next type in node sequence: {next}"
                     self.edge_index[source].append(Entry[T, S](next=next, config=Config(), index=i))
 
                 continue

src/edgygraph/graph/graphs.py

@@ -10,7 +10,7 @@
 from ..states import StateProtocol, SharedProtocol
 from ..diff import Change, ChangeConflictException, Diff
 from ..nodes import Node, END, START
-from .types import SingleNext, NextNode, ErrorEntry, SingleErrorSource, Entries, BranchContainer, SingleSource, Source, Types, ResolvedNext
+from .types import SingleNext, NextNode, ErrorEntry, SingleErrorSource, Entries, BranchContainer, SingleSource, Source, Types, ResolvedNext, Join
 from .hooks import GraphHook
 from .branches import Branch
 
@@ -35,7 +35,7 @@ class Graph[T: StateProtocol = StateProtocol, S: SharedProtocol = SharedProtocol
 
     For the more flexible approach with better scaling use protocols to define the supported state types. Remember to always extend `typing.Protocol` in the child classes for typing.
 
-    This is recommended for scalable projects where many different state types need to be joined in one graph. See https://github.com/mathisxy/edgynodes/ for an example.
+    This is recommended for scalable projects where many different state types need to be joined in one graph. See [edgynodes](https://github.com/mathisxy/edgynodes/) for an example.
 
     ### Disable Type Checking
 
@@ -46,13 +46,50 @@ class Graph[T: StateProtocol = StateProtocol, S: SharedProtocol = SharedProtocol
 
     The edges are defined as a list of tuples, where the first element is the source and the second element reveals the next node.
 
+    ### Branches
+
+    The edges are contained in branches. A branch is a tuple with edges and a join parameter at the end. 
+
+    #### Spawning
+
+    A branch is spawned when the source of the first edge of the branch is triggered.
+
+    In this example it would be on `START`:
+
+    ```python
+    edges=[(
+       (START, node1),
+       (node1, node2),
+       END
+    )]
+    ```
+
+    In this example it would be on `node1` and on `node2` each:
+
+    ```python
+    edges=[(
+       ([node1, node2], node3),
+       node4
+    )]
+    ```
+
+    #### Joining
+
+    The join parameter can be of the following types:
+
+    - `None`: The branch will not be joined.
+    - `END`: The branch will be joined at the end of the graph.
+    - A node instance: The branch will be joined directly before the given node is executed in any branch into this branch.
+
+    The process of joining describes waiting for the branches to finish wich aim to join and then applying the changes to the state of the whole finished branch to the state of the joining branch.
+
     ### Formats
 
-    The graph supports different formats for the edges.
+    A branch supports different formats for the edges.
 
     - `(source, target)`: A single edge from source to target.
     - `(START, target)`: A single edge from the start of the graph to target.
-    - `(source, END)`: A single edge from source to the end of the graph. It equals to `(source, None)`. It is redundant but can be used for better readability.
+    - `(source, None)`: A single edge from source to no target. `END` is not allowed here, since it would be redundant. It is only allowed in join parameters to distinct join at the end of the graph (`END`) from not joining (`None`)
     - `([source1, source2], target)`: Multiple edges from source1 and source2 to target.
     - `(source, [target1, target2])`: Multiple edges from source to target1 and target2.
     - `([source1, source2], [target1, target2])`: Multiple edges from source1 and source2 to target1 and target2. This will create 4 edges in total.
@@ -87,7 +124,7 @@ def __init__(self,
         self.hooks = hooks or []
 
         self.branch_registry: dict[SingleSource[T, S], list[Branch[T, S]]] = defaultdict(list)
-        self.join_registry: dict[SingleNext[T, S], list[Branch[T, S]]] = defaultdict(list)
+        self.join_registry: dict[Join[T, S], list[Branch[T, S]]] = defaultdict(list)
 
         self.index_branches()
 
@@ -257,14 +294,6 @@ async def node_wrapper(self, state: T, shared: S, node: NextNode[T, S]):
             raise e
 
 
-            
-    def spawn_branch(self, state: T, shared: S, branch: Branch[T, S]) -> None:
-
-        self.join_registry[branch.join].append(branch)
-
-        self.task_group.create_task(self.run_branch(state, shared, branch))
-
-
 
     async def merge_states(self, current_state: T, result_states: list[T]) -> T:
         """
@@ -356,6 +385,13 @@ async def spawn_branches(self, state: T, shared: S, next_nodes: list[NextNode[T,
 
                 for h in self.hooks: await h.on_spawn_branch_end(state, shared, branch, node, self.branch_registry, self.join_registry)
 
+    
+            
+    def spawn_branch(self, state: T, shared: S, branch: Branch[T, S]) -> None:
+
+        self.join_registry[branch.join].append(branch)
+
+        self.task_group.create_task(self.run_branch(state, shared, branch))
 
     async def join_branches(self, state: T, next_nodes: list[NextNode[T, S]]) -> T:
         """
@@ -569,11 +605,7 @@ def match(x: SingleNext[T, S]) -> None:
 
                 case None:
                     print("None")
-                    pass # END
-
-                case type():
-                    print("END")
-                    assert next is END, "Only END is allowed as a type here"
+                    pass
 
                 case Node():
                     print("Node")

src/edgygraph/graph/hooks.py

@@ -4,7 +4,7 @@
 from ..states import StateProtocol, SharedProtocol
 from ..diff import Change
 from .branches import Branch
-from .types import NextNode, SingleSource, SingleNext
+from .types import NextNode, SingleSource, Join
 
 
 
@@ -56,7 +56,7 @@ async def on_step_end(self, state: T, shared: S, nodes: list[NextNode[T, S]]) ->
         pass
 
 
-    async def on_spawn_branch_start(self, state: T, shared: S, branch: Branch[T, S], trigger: NextNode[T, S], branch_registry: dict[SingleSource[T, S], list[Branch[T, S]]], join_registry: dict[SingleNext[T, S], list[Branch[T, S]]]):
+    async def on_spawn_branch_start(self, state: T, shared: S, branch: Branch[T, S], trigger: NextNode[T, S], branch_registry: dict[SingleSource[T, S], list[Branch[T, S]]], join_registry: dict[Join[T, S], list[Branch[T, S]]]):
         """
         Called before a branch is spawned.
 
@@ -71,7 +71,7 @@ async def on_spawn_branch_start(self, state: T, shared: S, branch: Branch[T, S],
         pass
 
 
-    async def on_spawn_branch_end(self, state: T, shared: S, branch: Branch[T, S], trigger: NextNode[T, S], branch_registry: dict[SingleSource[T, S], list[Branch[T, S]]], join_registry: dict[SingleNext[T, S], list[Branch[T, S]]]):
+    async def on_spawn_branch_end(self, state: T, shared: S, branch: Branch[T, S], trigger: NextNode[T, S], branch_registry: dict[SingleSource[T, S], list[Branch[T, S]]], join_registry: dict[Join[T, S], list[Branch[T, S]]]):
         """
         Called after a branch is spawned.
 

src/edgygraph/graph/types.py

@@ -15,15 +15,16 @@
 type SingleErrorSource[T: StateProtocol, S: SharedProtocol] = type[Exception] | tuple[Node[T, S], type[Exception]]
 type ErrorSource[T: StateProtocol, S: SharedProtocol] = SingleErrorSource[T, S] | Sequence[SingleErrorSource[T, S]]
 
-type SingleNext[T: StateProtocol, S: SharedProtocol] = Node[T, S] | type[END] | None
+type SingleNext[T: StateProtocol, S: SharedProtocol] = Node[T, S] | None
 type ResolvedNext[T: StateProtocol, S: SharedProtocol] = SingleNext[T, S] | Sequence[SingleNext[T, S]]
 type Next[T: StateProtocol, S: SharedProtocol] = ResolvedNext[T, S] | Callable[[T, S], ResolvedNext[T, S]] | Callable[[T, S], Awaitable[ResolvedNext[T, S]]]
 
 
 type Edge[T: StateProtocol, S: SharedProtocol] = tuple[Source[T, S], Next[T, S]] | tuple[Source[T, S], Next[T, S], Config]
 type ErrorEdge[T: StateProtocol, S: SharedProtocol] = tuple[ErrorSource[T, S], Next[T, S]] | tuple[ErrorSource[T, S], Next[T, S], ErrorConfig]
 
-type BranchContainer[T: StateProtocol, S: SharedProtocol] = tuple[Edge[T, S] | NodeTupel[T, S], *tuple[Edge[T, S] | ErrorEdge[T, S] | NodeTupel[T, S], ...], SingleNext[T, S]]
+type Join[T: StateProtocol, S: SharedProtocol] = Next[T, S] | type[END]
+type BranchContainer[T: StateProtocol, S: SharedProtocol] = tuple[Edge[T, S] | NodeTupel[T, S], *tuple[Edge[T, S] | ErrorEdge[T, S] | NodeTupel[T, S], ...], Join[T, S]]
 
 
 class Types[T: StateProtocol, S: SharedProtocol]:

src/edgygraph/graph_hooks/node_print.py

@@ -2,7 +2,7 @@
 
 from ..graph.hooks import GraphHook
 from ..states import StateProtocol, SharedProtocol
-from ..graph.types import NextNode, SingleSource, SingleNext
+from ..graph.types import NextNode, SingleSource, Join
 from .utils.rich_printing import GraphRenderer
 
 class NodePrintHook[T: StateProtocol = StateProtocol, S: SharedProtocol = SharedProtocol](GraphHook[T, S]):
@@ -29,6 +29,6 @@ async def on_step_end(self, state: T, shared: S, nodes: list[NextNode[T, S]]) ->
     async def on_graph_end(self, state: T, shared: S) -> None:
         self.renderer.render_graph_end(state, shared)
 
-    async def on_spawn_branch_end(self, state: T, shared: S, branch: Branch[T, S], trigger: NextNode[T, S], branch_registry: dict[SingleSource[T, S], list[Branch[T, S]]], join_registry: dict[SingleNext[T, S], list[Branch[T, S]]]):
+    async def on_spawn_branch_end(self, state: T, shared: S, branch: Branch[T, S], trigger: NextNode[T, S], branch_registry: dict[SingleSource[T, S], list[Branch[T, S]]], join_registry: dict[Join[T, S], list[Branch[T, S]]]):
         self.renderer.render_spawn_branch_end(branch, trigger)
         self.renderer.render_branch_overview(branch_registry, join_registry)

src/edgygraph/graph_hooks/utils/rich_printing.py

@@ -9,7 +9,7 @@
 from rich.tree import Tree
 
 from ...diff import Change, ChangeTypes
-from ...graph.types import NextNode, SingleNext, SingleSource
+from ...graph.types import NextNode, Join, SingleSource
 from ...graph.branches import Branch
 from ...states import StateProtocol as State, SharedProtocol as Shared
 
@@ -225,7 +225,7 @@ def render_spawn_branch_end(
     def render_branch_overview(
         self,
         branch_registry: dict[SingleSource[T, S], list[Branch[T, S]]],
-        join_registry: dict[SingleNext[T, S], list[Branch[T, S]]],
+        join_registry: dict[Join[T, S], list[Branch[T, S]]],
     ) -> None:
         """
         Render a combined overview of branch_registry and join_registry.

tests/test_graph.py

@@ -174,7 +174,7 @@ def noop(self):
     def test_single_node_increments_value(self):
         state = SimpleState(value=0)
         shared = SimpleShared()
-        g = Graph[SimpleState, SimpleShared](edges=[((START, inc), (inc, END), END)])
+        g = Graph[SimpleState, SimpleShared](edges=[((START, inc), END)])
         result_state, _ = asyncio.get_event_loop().run_until_complete(g(state, shared))
         assert result_state.value == 1
 
@@ -183,14 +183,14 @@ def test_chain_of_two_nodes(self):
         n2 = IncrementNode()
         state = SimpleState(value=0)
         shared = SimpleShared()
-        g = Graph(edges=[((START, n1), (n1, n2), (n2, END), END)])
+        g = Graph(edges=[((START, n1), (n1, n2), (n2, None), END)])
         result_state, _ = asyncio.get_event_loop().run_until_complete(g(state, shared))
         assert result_state.value == 2
 
     def test_empty_graph_returns_unchanged_state(self):
         state = SimpleState(value=42)
         shared = SimpleShared()
-        g = Graph[SimpleState, SimpleShared](edges=[((START, END), END)])
+        g = Graph[SimpleState, SimpleShared](edges=[((START, None), END)])
         result_state, _ = asyncio.get_event_loop().run_until_complete(g(state, shared))
         assert result_state.value == 42
 
@@ -204,7 +204,7 @@ def test_no_edges_from_start_returns_unchanged(self):
     def test_shared_is_same_object(self):
         state = SimpleState()
         shared = SimpleShared()
-        g = Graph[SimpleState, SimpleShared](edges=[((START, inc), (inc, END), END)])
+        g = Graph[SimpleState, SimpleShared](edges=[((START, inc), END)])
         _, result_shared = asyncio.get_event_loop().run_until_complete(g(state, shared))
         assert result_shared is shared
 
@@ -219,7 +219,7 @@ def test_conditional_next_based_on_state(self):
         noop = NoOpNode()
 
         def router(state: SimpleState, shared: SimpleShared):
-            return noop if state.value > 0 else END
+            return noop if state.value > 0 else None
 
         state = SimpleState(value=1)
         shared = SimpleShared()
@@ -232,7 +232,7 @@ def test_conditional_returns_end(self):
         inc = IncrementNode()
 
         def router(state: SimpleState, shared: SimpleShared):
-            return END
+            return None
 
         state = SimpleState(value=0)
         shared = SimpleShared()
@@ -249,7 +249,7 @@ async def async_router(state: SimpleState, shared: SimpleShared):
 
         state = SimpleState(value=0)
         shared = SimpleShared()
-        g = Graph(edges=[((START, inc), (inc, async_router), (noop, END), END)])
+        g = Graph(edges=[((START, inc), (inc, async_router), END)])
         result_state, _ = asyncio.get_event_loop().run_until_complete(g(state, shared))
         assert result_state.value == 1
 
@@ -279,7 +279,6 @@ async def __call__(self, state: SimpleState, shared: SimpleShared) -> None:
         g = Graph(edges=[(
             (START, [sv, sn]),
             ([sv, sn], join),
-            (join, END),
             END)
         ])
         result_state, _ = asyncio.get_event_loop().run_until_complete(g(state, shared))
@@ -324,7 +323,6 @@ def test_error_edge_by_exception_type(self):
         g = Graph(edges=[(
             (START, raiser),
             (ValueError, recovery),
-            (recovery, END),
             END)
         ])
         result_state, _ = asyncio.get_event_loop().run_until_complete(g(state, shared))
@@ -339,7 +337,6 @@ def test_error_edge_by_node_and_exception_type(self):
         g = Graph(edges=[(
             (START, raiser),
             ((raiser, RuntimeError), recovery),
-            (recovery, END),
             END)
         ])
         result_state, _ = asyncio.get_event_loop().run_until_complete(g(state, shared))
@@ -387,7 +384,6 @@ def test_instant_node_runs_in_same_step(self):
         g = Graph(edges=[(
             (START, inc),
             (inc, noop, Config(instant=True)),
-            (noop, END),
             END)
         ])
         result_state, _ = asyncio.get_event_loop().run_until_complete(g(state, shared))
@@ -412,7 +408,6 @@ def test_list_source_registers_for_each_node(self):
             (START, [n1, n3]),
             (n3, n2),
             ([n1, n2], join),
-            (join, END),
             END)
         ])
         result_state, _ = asyncio.get_event_loop().run_until_complete(g(state, shared))