Graph and Execution

This section covers the graph representation and execution model.

ExecutableGraph

The ExecutableGraph represents a materialized flow ready for execution.

class flow.ExecutableGraph(name, nodes, connections, source_nodes)[source]

Bases: object

Executable graph created directly by FlowBuilder.

Parameters:

name (str)
nodes (Dict[str, Node])
connections (List[Connection[Any]])
source_nodes (List[Node])

__init__(name, nodes, connections, source_nodes)[source]

Parameters:

name (str)
nodes (Dict[str, Node])
connections (List[Connection[Any]])
source_nodes (List[Node])

async run(duration=None, auto_stop=True)[source]

Execute the graph.

Parameters:

duration (Optional[float], default: None)
auto_stop (bool, default: True)

Return type:

None

async stop()[source]

Stop graph execution.

Return type:: None

Properties

ExecutableGraph.name: The name of the graph.

ExecutableGraph.nodes: List of all nodes in the graph.

ExecutableGraph.connections: List of all connections between nodes.

ExecutableGraph.source_nodes: List of nodes with no input connections (entry points).

Methods

ExecutableGraph.run(duration=None, auto_stop=True)

Execute the graph asynchronously.

Parameters:

duration – Maximum duration in seconds (None for unlimited)
auto_stop – Stop when all sources complete (True) or run indefinitely (False)

Returns:

None

Example:

# 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)

ExecutableGraph.stop(mode=ShutdownMode.GRACEFUL)

Stop the graph execution.

Parameters:: mode – How to stop the graph
Returns:: None

Example:

# 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:

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:

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:

# 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:

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:

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:

# 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:

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

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