Merge pull request #1 from mdrideout/enforce-graph-builder-pattern

Enforce graph builder pattern

Author: mdrideout

.github/workflows/publish.yml

@@ -0,0 +1,31 @@
+name: Publish to PyPI
+
+on:
+  release:
+    types: [created]
+
+jobs:
+  publish:
+    runs-on: ubuntu-latest
+    environment:
+      name: pypi
+      url: https://pypi.org/project/junjo/
+    permissions:
+      id-token: write
+      contents: read
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.x'
+
+      - name: Build package
+        run: |
+          pip install build
+          python -m build
+
+      - name: Publish package distributions to PyPI
+        uses: pypa/gh-action-pypi-publish@release/v1

README.md

@@ -101,12 +101,12 @@ $ brew install graphviz
 
 ```python
 # visualize.py
-from base.sample_workflow.workflow import sample_workflow_graph
+from base.sample_workflow.graph import create_sample_workflow_graph
 
 def main():
     # Every graph can execute .export_graphviz_assets() to generate all graphs and subflow graphs in a workflow
     # Creates .svg renderings, .dot notation files, and an HTML template to render the graphs
-    sample_workflow_graph.export_graphviz_assets()
+    create_sample_workflow_graph().export_graphviz_assets()
 
 if __name__ == "__main__":
     main()
@@ -141,6 +141,13 @@ This project utilizes [ruff](https://astral.sh/ruff) for linting and auto format
 $ sphinx-build -b html docs docs/_build
 ```
 
+### Tests
+
+```bash
+# Run the tests with uv
+$ uv run pytest
+```
+
 ## Code Generation
 
 ### Protobuf schema generation

docs/.DS_Store

@@ -0,0 +1,162 @@
+.. _core_concepts:
+
+##############################################################
+Core Concepts
+##############################################################
+
+.. meta::
+    :description: Understand the core concepts of Junjo, including State, Store, Node, Edge, Condition, Graph, and Workflow. Learn how these components work together to build powerful and scalable Python workflows.
+    :keywords: junjo, python, workflow, state management, node, edge, graph, core concepts
+
+This page breaks down the fundamental building blocks of the Junjo library. Understanding these concepts is key to effectively designing, building, and debugging your workflows.
+
+State
+=====
+
+**What is it?**
+A `BaseState` is a Pydantic model that defines the data structure for your workflow's state. It acts as a centralized, type-safe container for all the data that your workflow will operate on.
+
+**Key Characteristics:**
+- **Pydantic-Based:** Leverages Pydantic for data validation and type hinting.
+- **Immutable in Practice:** While the state object itself can be replaced, it is treated as immutable within the workflow. Nodes do not modify the state directly; they request changes through the store.
+
+.. code-block:: python
+
+    from junjo import BaseState
+
+    class MyWorkflowState(BaseState):
+        user_input: str
+        processed_data: dict | None = None
+        is_complete: bool = False
+
+Store
+=====
+
+**What is it?**
+A `BaseStore` is a class that manages the state of a workflow. It holds the `BaseState` and provides methods (often called "actions") to update the state in a controlled and predictable manner.
+
+**Key Characteristics:**
+- **State Management:** The single source of truth for the workflow's state.
+- **Redux-Inspired:** Follows a pattern where state is updated by dispatching actions, ensuring that state changes are explicit and traceable.
+- **Concurrency Safe:** Uses an `asyncio.Lock` to ensure that state updates are atomic, preventing race conditions.
+
+.. code-block:: python
+
+    from junjo import BaseStore
+
+    class MyWorkflowStore(BaseStore[MyWorkflowState]):
+        async def set_processed_data(self, data: dict) -> None:
+            await self.set_state({"processed_data": data})
+
+        async def mark_as_complete(self) -> None:
+            await self.set_state({"is_complete": True})
+
+Node
+====
+
+**What is it?**
+A `Node` represents a single unit of work in your workflow. It's where your business logic, API calls, or any other operations are executed.
+
+**Key Characteristics:**
+- **Atomic Unit of Work:** Each node should have a single, well-defined responsibility.
+- **Interacts with the Store:** Nodes receive the workflow's store as an argument to their `service` method, allowing them to read the current state and dispatch actions to update it.
+- **Asynchronous:** The `service` method is an `async` function, allowing for non-blocking I/O operations.
+
+.. code-block:: python
+
+    from junjo import Node
+
+    class ProcessDataNode(Node[MyWorkflowStore]):
+        async def service(self, store: MyWorkflowStore) -> None:
+            state = await store.get_state()
+            # Perform some processing on state.user_input
+            processed_data = {"result": "some_value"}
+            await store.set_processed_data(processed_data)
+
+Edge
+====
+
+**What is it?**
+An `Edge` defines a directed connection between two nodes in a workflow graph. It represents a potential path of execution.
+
+**Key Characteristics:**
+- **Defines Flow:** Edges connect a `tail` node to a `head` node, establishing the sequence of operations.
+- **Can be Conditional:** An edge can have an associated `Condition` that determines whether the transition from the tail to the head should occur.
+
+.. code-block:: python
+
+    from junjo import Edge
+
+    edge = Edge(tail=node1, head=node2)
+
+Condition
+=========
+
+**What is it?**
+A `Condition` is a class that contains logic to determine whether an `Edge` should be traversed.
+
+**Key Characteristics:**
+- **Pure Function of State:** A condition's `evaluate` method should only depend on the current state of the workflow. It should not have any side effects.
+- **Enables Branching:** Conditions are the primary mechanism for creating branching logic in your workflows.
+
+.. code-block:: python
+
+    from junjo import Condition
+
+    class DataIsProcessed(Condition[MyWorkflowState]):
+        def evaluate(self, state: MyWorkflowState) -> bool:
+            return state.processed_data is not None
+
+    edge = Edge(tail=node1, head=node2, condition=DataIsProcessed())
+
+Graph
+=====
+
+**What is it?**
+A `Graph` is a collection of nodes and edges that defines the complete structure of your workflow.
+
+**Key Characteristics:**
+- **Source and Sink:** A graph has a single entry point (`source`) and a single exit point (`sink`).
+- **Defines the Workflow Structure:** The graph is a complete representation of all possible paths of execution in your workflow.
+
+.. code-block:: python
+
+    from junjo import Graph
+
+    workflow_graph = Graph(
+        source=start_node,
+        sink=end_node,
+        edges=[
+            Edge(tail=start_node, head=process_node),
+            Edge(tail=process_node, head=end_node, condition=DataIsProcessed())
+        ]
+    )
+
+Workflow
+========
+
+**What is it?**
+A `Workflow` is the main executable component that takes a `graph_factory` and a `store_factory` and runs the defined process.
+
+**Key Characteristics:**
+- **Executable:** The `Workflow` class has an `execute` method that starts the workflow.
+- **Manages Execution:** It traverses the graph, executing nodes and evaluating conditions, until the `sink` node is reached.
+- **Isolated Execution:** Each call to `execute` uses the provided factories to create a fresh `Graph` and `Store`, ensuring that each execution is isolated and concurrency-safe.
+
+.. code-block:: python
+
+    from junjo import Workflow
+
+    def create_graph() -> Graph:
+        # ... (graph creation logic)
+        return workflow_graph
+
+    sample_workflow = Workflow[MyWorkflowState, MyWorkflowStore](
+        name="My First Workflow",
+        graph_factory=create_graph,
+        store_factory=lambda: MyWorkflowStore(
+            initial_state=MyWorkflowState(user_input="hello")
+        )
+    )
+
+    await sample_workflow.execute()

docs/core_concepts.rst

@@ -87,34 +87,40 @@ More advanced examples can be found in the `examples directory <https://github.c
                     return False
                 return count % 2 == 0
 
-        # Instantiate the nodes
-        first_node = FirstNode()
-        count_items_node = CountItemsNode()
-        even_items_node = EvenItemsNode()
-        odd_items_node = OddItemsNode()
-        final_node = FinalNode()
-
-        # Create the workflow graph
-        workflow_graph = Graph(
-            source=first_node,
-            sink=final_node,
-            edges=[
-                Edge(tail=first_node, head=count_items_node),
-
-                # Branching based on the count of items
-                Edge(tail=count_items_node, head=even_items_node, condition=CountIsEven()), # Only transitions if count is even
-                Edge(tail=count_items_node, head=odd_items_node), # Fallback if first condition is not met
-
-                # Branched paths converge to the final node
-                Edge(tail=even_items_node, head=final_node),
-                Edge(tail=odd_items_node, head=final_node),
-            ]
-        )
+        def create_graph() -> Graph:
+            """
+            Factory function to create a new instance of the sample workflow graph.
+            This ensures that each workflow execution gets a fresh, isolated graph,
+            preventing state conflicts in concurrent environments.
+            """
+            # Instantiate the nodes
+            first_node = FirstNode()
+            count_items_node = CountItemsNode()
+            even_items_node = EvenItemsNode()
+            odd_items_node = OddItemsNode()
+            final_node = FinalNode()
+
+            # Create the workflow graph
+            return Graph(
+                source=first_node,
+                sink=final_node,
+                edges=[
+                    Edge(tail=first_node, head=count_items_node),
+
+                    # Branching based on the count of items
+                    Edge(tail=count_items_node, head=even_items_node, condition=CountIsEven()), # Only transitions if count is even
+                    Edge(tail=count_items_node, head=odd_items_node), # Fallback if first condition is not met
+
+                    # Branched paths converge to the final node
+                    Edge(tail=even_items_node, head=final_node),
+                    Edge(tail=odd_items_node, head=final_node),
+                ]
+            )
 
         # Create the workflow
         sample_workflow = Workflow[SampleWorkflowState, SampleWorkflowStore](
             name="Getting Started Example Workflow",
-            graph=workflow_graph,
+            graph_factory=create_graph,
             store_factory=lambda: SampleWorkflowStore(
                 initial_state=SampleWorkflowState(
                     items=["laser", "coffee", "horse"]

docs/getting_started.rst

@@ -4,6 +4,9 @@
    :hidden:
 
    getting_started
+   tutorial
+   core_concepts
+   state_management
    concurrency
    subflows
    visualizing_workflows

docs/index.rst

@@ -0,0 +1,87 @@
+.. _state_management:
+
+##############################################################
+State Management
+##############################################################
+
+.. meta::
+    :description: A deep dive into Junjo's Redux-inspired, immutable state management. Learn how to use BaseState and BaseStore to build predictable and concurrency-safe Python workflows.
+    :keywords: junjo, python, state management, redux, immutable state, pydantic, workflow, BaseStore, BaseState
+
+Junjo's state management is designed to be predictable, traceable, and safe for concurrent operations. It is heavily inspired by the principles of Redux, a popular state management library in the JavaScript ecosystem. This page provides a deep dive into how to effectively manage state in your Junjo workflows.
+
+The Core Principles
+===================
+
+1.  **Single Source of Truth:** The state of your entire workflow is stored in a single object tree within a single **Store**.
+2.  **State is Read-Only:** The only way to change the state is to emit an "action," an object describing what happened. This prevents nodes from directly modifying the state, which could lead to unpredictable behavior.
+3.  **Changes are Made with Store Methods:** State modifications are encapsulated within methods in your **Store**. Similar to "reducers" in Redux, these methods are the only place where `set_state` should be called, ensuring that all state changes are predictable and centralized.
+
+BaseState: Defining Your State's Shape
+=======================================
+
+The `BaseState` class, which is a Pydantic `BaseModel`, is used to define the structure of your workflow's state. Because it's a Pydantic model, you get all the benefits of type hinting and data validation out of the box.
+
+.. code-block:: python
+
+    from junjo import BaseState
+
+    class ChatWorkflowState(BaseState):
+        messages: list[dict] = []
+        current_user: str
+        is_typing: bool = False
+        error_message: str | None = None
+
+In this example, we've defined a state for a chat application. Any workflow that uses this state will have access to these fields, and Pydantic will ensure that the data conforms to the specified types.
+
+BaseStore: Managing Your State
+==============================
+
+The `BaseStore` is the heart of Junjo's state management. It holds the state and provides methods for updating it. You will create a custom store for each workflow that inherits from `BaseStore` and is typed with your custom `BaseState`.
+
+.. code-block:: python
+
+    from junjo import BaseStore
+
+    class ChatWorkflowStore(BaseStore[ChatWorkflowState]):
+        async def add_message(self, message: dict) -> None:
+            # Get the current messages and append the new one
+            new_messages = self._state.messages + [message]
+            await self.set_state({"messages": new_messages})
+
+        async def set_is_typing(self, is_typing: bool) -> None:
+            await self.set_state({"is_typing": is_typing})
+
+        async def set_error(self, error: str) -> None:
+            await self.set_state({"error_message": error})
+
+### The `set_state` Method
+
+The `set_state` method is the **only** way to update the state in the store. It takes a dictionary of the fields you want to update and their new values.
+
+**Key Behaviors of `set_state`:**
+- **Immutable Updates:** `set_state` creates a *copy* of the state with the updates applied. It does not mutate the original state object. This is crucial for preventing side effects and ensuring predictable state transitions.
+- **Concurrency-Safe:** All calls to `set_state` are protected by an `asyncio.Lock`, so you can safely call actions from multiple concurrent nodes without worrying about race conditions.
+- **Validation:** Before applying the update, `set_state` validates the new state against your Pydantic model. If the update is invalid, it will raise a `ValueError`.
+
+Using the Store in a Node
+=========================
+
+Nodes receive an instance of the store in their `service` method. This allows them to read the current state and dispatch actions to update it.
+
+.. code-block:: python
+
+    from junjo import Node
+
+    class SendMessageNode(Node[ChatWorkflowStore]):
+        async def service(self, store: ChatWorkflowStore) -> None:
+            state = await store.get_state()
+            user = state.current_user
+            
+            # In a real app, you would get the message from an external source
+            new_message = {"user": user, "text": "Hello, Junjo!"}
+
+            # Dispatch an action to add the message to the state
+            await store.add_message(new_message)
+
+By following this pattern, you create a clear and predictable data flow in your application. Nodes don't need to know how the state is updated; they just need to know which actions to call on the store. This separation of concerns makes your code easier to test, debug, and reason about.