Graph and Execution
===================

This section covers the graph representation and execution model.

ExecutableGraph
---------------

The ExecutableGraph represents a materialized flow ready for execution.

.. autoclass:: flow.ExecutableGraph
   :members:
   :special-members: __init__
   :show-inheritance:

Properties
~~~~~~~~~~

.. attribute:: ExecutableGraph.name

   The name of the graph.

.. attribute:: ExecutableGraph.nodes

   List of all nodes in the graph.

.. attribute:: ExecutableGraph.connections

   List of all connections between nodes.

.. attribute:: ExecutableGraph.source_nodes

   List of nodes with no input connections (entry points).

Methods
~~~~~~~

.. method:: ExecutableGraph.run(duration=None, auto_stop=True)

   Execute the graph asynchronously.
   
   :param duration: Maximum duration in seconds (None for unlimited)
   :param auto_stop: Stop when all sources complete (True) or run indefinitely (False)
   :return: None
   
   Example:
   
   .. code-block:: python
   
      # Run until sources complete
      await graph.run()
      
      # Run for specific duration
      await graph.run(duration=10.0)
      
      # Run as server (manual stop required)
      await graph.run(auto_stop=False)

.. method:: ExecutableGraph.stop(mode=ShutdownMode.GRACEFUL)

   Stop the graph execution.
   
   :param mode: How to stop the graph
   :return: None
   
   Example:
   
   .. code-block:: python
   
      # Graceful shutdown (default)
      await graph.stop()
      
      # Immediate shutdown
      await graph.stop(ShutdownMode.IMMEDIATE)
      
      # Force shutdown
      await graph.stop(ShutdownMode.FORCE)

Graph Patterns
--------------

Pipeline Pattern
~~~~~~~~~~~~~~~~

Process a finite dataset from start to finish:

.. code-block:: python

   from flow import flow
   
   # Pipeline with auto_stop=True (default)
   graph = await (
       flow("Data Pipeline")
       .source(dataset, dict)
       .transform(process, dict)
       .to(save_result)
       .execute(auto_stop=True)
   )

Server Pattern
~~~~~~~~~~~~~~

Continuous processing with manual control:

.. code-block:: python

   from flow import flow
   import asyncio
   
   # Start server
   graph = await (
       flow("API Server")
       .source(request_stream(), dict)
       .transform(handle_request, dict)
       .to(send_response)
       .execute(auto_stop=False)
   )
   
   # Run until shutdown signal
   try:
       await shutdown_event.wait()
   finally:
       await graph.stop()

Concurrent Execution
~~~~~~~~~~~~~~~~~~~~

Multiple graphs can run concurrently:

.. code-block:: python

   # Create multiple graphs
   graph1 = flow("Pipeline 1").source(data1, int).to(results1.append).build()
   graph2 = flow("Pipeline 2").source(data2, int).to(results2.append).build()
   
   # Run concurrently
   await asyncio.gather(
       graph1.run(),
       graph2.run()
   )

Error Handling
--------------

Graphs handle errors at multiple levels:

Node-Level Errors
~~~~~~~~~~~~~~~~~

Errors in individual nodes trigger the ``on_error`` lifecycle hook:

.. code-block:: python

   class ErrorHandlingNode(Node):
       async def on_error(self, error: Exception):
           logger.error(f"Error in {self.name}: {error}")
           # Optionally re-raise to propagate
           raise

Graph-Level Errors
~~~~~~~~~~~~~~~~~~

Unhandled errors will stop the graph:

.. code-block:: python

   try:
       await graph.run()
   except Exception as e:
       logger.error(f"Graph failed: {e}")
       # Graph automatically stops on unhandled errors

Monitoring
----------

Runtime Inspection
~~~~~~~~~~~~~~~~~~

Inspect graph state during execution:

.. code-block:: python

   # Check if graph is running
   if graph._running:
       print("Graph is active")
   
   # Access nodes
   for node in graph.nodes:
       print(f"Node: {node.name}, Type: {node.__class__.__name__}")
   
   # Check connections
   for conn in graph.connections:
       print(f"Connection: {conn.source.node.name} -> {conn.target.node.name}")

Metrics Collection
~~~~~~~~~~~~~~~~~~

Use MetricsMiddleware for detailed statistics:

.. code-block:: python

   from flow import flow, MetricsMiddleware
   
   metrics = MetricsMiddleware()
   
   graph = await (
       flow("Monitored Pipeline")
       .with_middleware(metrics)
       .source(data, int)
       .transform(process, int)
       .to(save)
       .execute()
   )
   
   # Get metrics after execution
   stats = metrics.get_metrics()
   print(f"Total processed: {stats['total_processed']}")
   print(f"Total errors: {stats['total_errors']}")

Best Practices
--------------

1. **Resource Management**: Always stop graphs explicitly in server mode
2. **Error Handling**: Implement ``on_error`` for recovery logic
3. **Monitoring**: Use middleware for production monitoring
4. **Testing**: Test graphs with finite data before production
5. **Shutdown**: Use graceful shutdown for data integrity