FlowBuilder API

The FlowBuilder provides a fluent interface for constructing data flows.

Creating a Flow

Use the flow.flow() function to create a new FlowBuilder:

from flow import flow

# Create a new flow
builder = flow("My Pipeline")

# Create with namespace
builder = flow("pipeline:data_processor")

FlowBuilder Methods

Source Methods

FlowBuilder.source(data, data_type, name=None)

Add a source node to the flow.

Parameters:

data – The data source (iterable, callable, or async iterator)
data_type – The type of data produced
name – Optional name for the node

Returns:

FlowBuilder for chaining

Example:

# From list
builder.source([1, 2, 3], int)

# From generator
def generate():
    for i in range(10):
        yield i

builder.source(generate(), int)

# From async source
async def fetch_data():
    async for item in api_stream():
        yield item

builder.source(fetch_data(), dict)

Transform Methods

FlowBuilder.transform(func, output_type, name=None)

Add a transformation to the flow.

Parameters:

func – Function to transform each item
output_type – The type of data produced
name – Optional name for the node

Returns:

FlowBuilder for chaining

Example:

# Simple transform
builder.transform(lambda x: x * 2, int)

# Named function
def process_data(item):
    return {"value": item, "processed": True}

builder.transform(process_data, dict)

Filter Methods

FlowBuilder.filter(predicate, name=None)

Filter items in the flow.

Parameters:

predicate – Function returning True to keep items
name – Optional name for the node

Returns:

FlowBuilder for chaining

Example:

# Filter even numbers
builder.filter(lambda x: x % 2 == 0)

# Complex filter
def is_valid(item):
    return item["status"] == "active" and item["value"] > 0

builder.filter(is_valid)

Side Effects

FlowBuilder.tap(func, name=None)

Add a side effect without transforming the data.

Parameters:

func – Function to execute for side effects
name – Optional name for the node

Returns:

FlowBuilder for chaining

Example:

# Logging
builder.tap(lambda x: print(f"Processing: {x}"))

# Metrics
metrics = []
builder.tap(metrics.append)

Sink Methods

FlowBuilder.sink(func, name=None)

FlowBuilder.to(func, name=None)

Add a sink node to consume data. to is an alias for sink.

Parameters:

func – Function to consume each item
name – Optional name for the node

Returns:

FlowBuilder for chaining

Example:

# Collect results
results = []
builder.to(results.append)

# Write to file
async def write_to_file(item):
    async with aiofiles.open("output.txt", "a") as f:
        await f.write(f"{item}\n")

builder.to(write_to_file)

Flow Control

FlowBuilder.split(count=2, name=None)

Split the flow into multiple outputs.

Parameters:

count – Number of output flows
name – Optional name for the node

Returns:

List of FlowBuilders

Example:

# Split into two streams
flow1, flow2 = builder.split(2)

# Process each stream differently
flow1.transform(lambda x: x * 2, int).to(results1.append)
flow2.filter(lambda x: x > 5).to(results2.append)

FlowBuilder.merge_with(*others, name=None)

Merge multiple flows together.

Parameters:

others – Other FlowBuilders to merge
name – Optional name for the node

Returns:

FlowBuilder for chaining

Example:

# Create separate flows
flow1 = flow("source1").source(data1, int)
flow2 = flow("source2").source(data2, int)

# Merge them
flow1.merge_with(flow2).to(results.append)

Middleware

FlowBuilder.with_middleware(*middleware)

Add middleware to all nodes in the flow.

Parameters:: middleware – Middleware instances to add
Returns:: FlowBuilder for chaining

Example:

from flow import LoggingMiddleware, RetryMiddleware

logger = LoggingMiddleware()
retry = RetryMiddleware(max_attempts=3)

builder.with_middleware(logger, retry)

Execution

FlowBuilder.build()

Build the flow into an ExecutableGraph.

Returns:: ExecutableGraph instance

Example:

graph = builder.build()
await graph.run()

FlowBuilder.execute(duration=None, auto_stop=None)

Build and execute the flow.

Parameters:

duration – Maximum duration in seconds
auto_stop – Whether to stop when sources complete

Returns:

ExecutableGraph instance

Example:

# Run for 10 seconds
await builder.execute(duration=10.0)

# Run until completion
await builder.execute(auto_stop=True)

# Run indefinitely (server mode)
graph = await builder.execute(auto_stop=False)
# ... later ...
await graph.stop()

Complete Example

from flow import flow, LoggingMiddleware
import asyncio

async def main():
    results = []

    # Build a complete pipeline
    await (
        flow("ETL Pipeline")
        .with_middleware(LoggingMiddleware())
        .source(range(1, 11), int)
        .filter(lambda x: x % 2 == 0)
        .transform(lambda x: x ** 2, int)
        .tap(lambda x: print(f"Squared: {x}"))
        .to(results.append)
        .execute()
    )

    print(f"Results: {results}")

asyncio.run(main())