Core API

This section documents the core Flokkit Flow API.

Flow Builder

The primary entry point for creating flows:

flow.flow(namespace=None)[source]

Create a new flow builder.

Parameters:

namespace (Optional[str], default: None) – Optional namespace for the flow (defaults to “flow”)

Return type:

FlowBuilder[Any]

Returns:

A new FlowBuilder instance

Examples

Create a simple flow:

>>> builder = flow()
>>> builder.namespace
'flow'

Create a named flow:

>>> builder = flow("DataPipeline")
>>> builder.namespace
'DataPipeline'

Complete pipeline example:

>>> import asyncio
>>> async def process_data():
...     results = []
...     await (
...         flow("Example")
...         .source(range(5), int)
...         .filter(lambda x: x % 2 == 0)
...         .transform(lambda x: x**2, int)
...         .sink(results.append)
...         .execute(duration=0.5)
...     )
...     return results
>>> asyncio.run(process_data())
[0, 4, 16]
class flow.FlowBuilder(namespace=None, current_node=None, current_port_name='out', current_port_type=None)[source]

Bases: Generic[T]

Builder for creating flow graphs with fluent interface.

Examples

Basic pipeline:

>>> import asyncio
>>> from flow import flow
>>> async def example():
...     results = []
...     await (
...         flow()
...         .source([1, 2, 3], int)
...         .transform(lambda x: x * 2, int)
...         .filter(lambda x: x > 2, int)
...         .sink(results.append)
...         .execute(duration=0.5)
...     )
...     return results
>>> asyncio.run(example())
[4, 6]

Using tap for side effects:

>>> async def example_tap():
...     tapped = []
...     results = []
...     await (
...         flow()
...         .source([1, 2, 3], int)
...         .tap(tapped.append)  # Side effect without consuming
...         .transform(lambda x: x * 10, int)
...         .sink(results.append)
...         .execute(duration=0.5)
...     )
...     return tapped, results
>>> tapped, results = asyncio.run(example_tap())
>>> tapped
[1, 2, 3]
>>> results
[10, 20, 30]
Parameters:
__init__(namespace=None, current_node=None, current_port_name='out', current_port_type=None)[source]
Parameters:
source(source, output_type, name=None)[source]

Add source node.

Parameters:

  • source (Any) – Data source - can be list, generator, iterator, or async iterator

  • output_type (Type[TypeVar(R)]) – Type of items produced by the source

  • name (Optional[str], default: None) – Optional debug name for the node

Return type:

FlowBuilder[TypeVar(R)]

Returns:

New FlowBuilder instance for chaining

Examples

From a list:

>>> builder = flow().source([1, 2, 3], int)
>>> builder._current_port_type
<class 'int'>

From a generator:

>>> def gen():
...     yield 1
...     yield 2
>>> builder = flow().source(gen(), int)

From a lambda:

>>> builder = flow().source(lambda: range(5), int)

From an async generator:

>>> async def async_gen():
...     for i in range(3):
...         yield i
>>> builder = flow().source(async_gen(), int)
transform(func, output_type, name=None, queue_size=1000, full_strategy=QueueFullStrategy.BLOCK)[source]

Add transform node.

Parameters:

  • func (Callable[[TypeVar(T)], TypeVar(R)]) – Function to transform each item

  • output_type (Type[TypeVar(R)]) – Type of transformed items

  • name (Optional[str], default: None) – Optional debug name for the node

  • queue_size (int, default: 1000) – Size of the input queue

  • full_strategy (QueueFullStrategy, default: <QueueFullStrategy.BLOCK: 1>) – Strategy when queue is full

Return type:

FlowBuilder[TypeVar(R)]

Returns:

New FlowBuilder instance for chaining

Examples

Simple transformation:

>>> import asyncio
>>> async def example():
...     results = []
...     await (
...         flow()
...         .source([1, 2, 3], int)
...         .transform(lambda x: x * 2, int)
...         .sink(results.append)
...         .execute(duration=0.5)
...     )
...     return results
>>> asyncio.run(example())
[2, 4, 6]

Async transformation:

>>> async def async_double(x):
...     await asyncio.sleep(0.001)  # Simulate async work
...     return x * 2
>>> async def example_async():
...     results = []
...     await (
...         flow()
...         .source([1, 2], int)
...         .transform(async_double, int)
...         .sink(results.append)
...         .execute(duration=0.5)
...     )
...     return results
>>> asyncio.run(example_async())
[2, 4]
filter(predicate, name=None, queue_size=1000, full_strategy=QueueFullStrategy.BLOCK)[source]

Add filter node.

Parameters:

  • predicate (Callable[[TypeVar(T)], bool]) – Function that returns True to keep items

  • name (Optional[str], default: None) – Optional debug name for the node

  • queue_size (int, default: 1000) – Size of the input queue

  • full_strategy (QueueFullStrategy, default: <QueueFullStrategy.BLOCK: 1>) – Strategy when queue is full

Return type:

FlowBuilder[TypeVar(T)]

Returns:

New FlowBuilder instance for chaining

Examples

Filter even numbers:

>>> import asyncio
>>> async def example():
...     results = []
...     await (
...         flow()
...         .source(range(6), int)
...         .filter(lambda x: x % 2 == 0)
...         .sink(results.append)
...         .execute(duration=0.5)
...     )
...     return results
>>> asyncio.run(example())
[0, 2, 4]

Filter with async predicate:

>>> async def is_valid(x):
...     await asyncio.sleep(0.001)  # Simulate async check
...     return x > 10
>>> async def example_async():
...     results = []
...     await (
...         flow()
...         .source([5, 15, 25], int)
...         .filter(is_valid)
...         .sink(results.append)
...         .execute(duration=0.5)
...     )
...     return results
>>> asyncio.run(example_async())
[15, 25]
tap(func, name=None, queue_size=1000, full_strategy=QueueFullStrategy.BLOCK)[source]

Add tap for side effects without consuming the stream.

Parameters:

  • func (Callable[[TypeVar(T)], None]) – Function to call for side effects (return value ignored)

  • name (Optional[str], default: None) – Optional debug name for the node

  • queue_size (int, default: 1000) – Size of the input queue

  • full_strategy (QueueFullStrategy, default: <QueueFullStrategy.BLOCK: 1>) – Strategy when queue is full

Return type:

FlowBuilder[TypeVar(T)]

Returns:

New FlowBuilder instance for chaining

Examples

Logging without modifying the stream:

>>> import asyncio
>>> async def example():
...     logged = []
...     results = []
...     await (
...         flow()
...         .source([1, 2, 3], int)
...         .tap(lambda x: logged.append(f"Processing {x}"))
...         .transform(lambda x: x**2, int)
...         .sink(results.append)
...         .execute(duration=0.5)
...     )
...     return logged, results
>>> logged, results = asyncio.run(example())
>>> logged
['Processing 1', 'Processing 2', 'Processing 3']
>>> results
[1, 4, 9]
sink(func, name=None, queue_size=1000, full_strategy=QueueFullStrategy.BLOCK)[source]

Add sink node.

Parameters:

  • func (Callable[[TypeVar(T)], Any]) – Function to consume each item

  • name (Optional[str], default: None) – Optional debug name for the node

  • queue_size (int, default: 1000) – Size of the input queue

  • full_strategy (QueueFullStrategy, default: <QueueFullStrategy.BLOCK: 1>) – Strategy when queue is full

Return type:

FlowBuilder[None]

Returns:

New FlowBuilder instance (cannot chain further transforms)

Examples

Collect to a list:

>>> import asyncio
>>> async def example():
...     results = []
...     await (
...         flow()
...         .source([1, 2, 3], int)
...         .sink(results.append)
...         .execute(duration=0.5)
...     )
...     return results
>>> asyncio.run(example())
[1, 2, 3]

Print items:

>>> async def example_print():
...     await (
...         flow()
...         .source(["hello", "world"], str)
...         .sink(print)
...         .execute(duration=0.5)
...     )
>>> # asyncio.run(example_print())  # Would print: hello\nworld
to(func, name=None, queue_size=1000, full_strategy=QueueFullStrategy.BLOCK)[source]

Alias for sink().

Examples

>>> import asyncio
>>> async def example():
...     results = []
...     await (
...         flow()
...         .source([1, 2, 3], int)
...         .to(results.append)  # Same as .sink()
...         .execute(duration=0.5)
...     )
...     return results
>>> asyncio.run(example())
[1, 2, 3]
Parameters:
Return type:

FlowBuilder[None]

split(n=2, queue_size=1000, full_strategy=QueueFullStrategy.BLOCK)[source]

Split stream into multiple outputs.

Parameters:

  • n (int, default: 2) – Number of output streams to create

  • queue_size (int, default: 1000) – Size of the input queue

  • full_strategy (QueueFullStrategy, default: <QueueFullStrategy.BLOCK: 1>) – Strategy when queue is full

Return type:

List[FlowBuilder[TypeVar(T)]]

Returns:

List of FlowBuilder instances, one for each output

Examples

Split and process separately:

>>> source = flow().source([1, 2, 3], int)
>>> streams = source.split(2)
>>> len(streams)
2
>>> # Each stream can be processed independently
>>> stream1, stream2 = streams
>>> stream1._current_port_name
'out0'
>>> stream2._current_port_name
'out1'
merge_with(*others, name=None, queue_size=1000, full_strategy=QueueFullStrategy.BLOCK)[source]

Merge multiple flows together.

Parameters:

  • *others (FlowBuilder[TypeVar(T)]) – Other FlowBuilder instances to merge with

  • name (Optional[str], default: None) – Optional debug name for the merge node

  • queue_size (int, default: 1000) – Size of the input queues

  • full_strategy (QueueFullStrategy, default: <QueueFullStrategy.BLOCK: 1>) – Strategy when queue is full

Return type:

FlowBuilder[TypeVar(T)]

Returns:

New FlowBuilder instance with merged streams

Examples

Merge two sources:

>>> import asyncio
>>> async def example():
...     results = []
...
...     source1 = flow().source([1, 2, 3], int)
...     source2 = flow().source([4, 5, 6], int)
...
...     await (
...         source1.merge_with(source2)
...         .sink(results.append)
...         .execute(duration=0.5)
...     )
...     return sorted(results)  # Sort for deterministic output
>>> asyncio.run(example())
[1, 2, 3, 4, 5, 6]
with_middleware(*middlewares)[source]

Add middleware to all nodes in the current flow.

Parameters:

*middlewares (Middleware) – Middleware instances to add

Return type:

FlowBuilder[TypeVar(T)]

Returns:

New FlowBuilder instance with middleware applied

Examples

Add logging middleware:

>>> import asyncio
>>> from flow import LoggingMiddleware
>>> async def example():
...     results = []
...     logger = LoggingMiddleware(log_inputs=False, log_outputs=False)
...
...     await (
...         flow()
...         .with_middleware(logger)
...         .source([1, 2], int)
...         .transform(lambda x: x * 2, int)
...         .sink(results.append)
...         .execute(duration=0.5)
...     )
...     return results
>>> asyncio.run(example())
[2, 4]
build()[source]

Build the executable graph.

Return type:

ExecutableGraph

Returns:

ExecutableGraph ready for execution

Examples

Build and inspect graph:

>>> from flow import flow
>>> builder = (
...     flow("MyPipeline")
...     .source([1, 2, 3], int)
...     .transform(lambda x: x * 2, int)
...     .sink(lambda x: None)
... )
>>> graph = builder.build()
>>> graph.name
'MyPipeline'
>>> len(graph.nodes)
3
async execute(duration=None, auto_stop=True)[source]

Build and execute the graph.

Parameters:

  • duration (Optional[float], default: None) – Maximum execution time in seconds

  • auto_stop (bool, default: True) – Whether to stop when all data is processed

Return type:

None

Examples

Execute a simple pipeline:

>>> import asyncio
>>> async def example():
...     results = []
...     await (
...         flow()
...         .source([1, 2, 3], int)
...         .transform(lambda x: x + 10, int)
...         .sink(results.append)
...         .execute(duration=0.5)
...     )
...     return results
>>> asyncio.run(example())
[11, 12, 13]

Core Components

Node Base Class

class flow.Node(name)[source]

Bases: ABC

Base class for all FBP nodes.

Parameters:

name (str)

__init__(name)[source]
Parameters:

name (str)

abstractmethod async process()[source]

Process data. Must be implemented by subclasses.

Return type:

None

add_input_port(name, data_type, queue_size=1000, full_strategy=QueueFullStrategy.BLOCK)[source]

Add an input port to this node.

Parameters:
Return type:

Port[TypeVar(T)]

add_output_port(name, data_type)[source]

Add an output port to this node.

Parameters:
Return type:

Port[TypeVar(T)]

get_input_port(name)[source]

Get an input port by name.

Parameters:

name (str)

Return type:

Port[Any]

get_output_port(name)[source]

Get an output port by name.

Parameters:

name (str)

Return type:

Port[Any]

async initialize()[source]

Initialize all ports and perform setup. Can be overridden.

Return type:

None

async on_start()[source]

Called when node starts. Override for custom initialization.

Return type:

None

async on_stop()[source]

Called when node stops. Override for cleanup.

Return type:

None

async on_error(error)[source]

Called when an error occurs. Override for custom error handling.

Parameters:

error (Exception)

Return type:

None

async on_complete()[source]

Called when node completes all processing. Override for custom completion logic.

Return type:

None

signal_completion()[source]

Signal that this node has completed and won’t produce more data.

Return type:

None

Port

class flow.Port(name, is_input, data_type, node=None, queue_size=1000, full_strategy=QueueFullStrategy.BLOCK)[source]

Bases: Generic[T]

Represents an input or output port with type information.

Parameters:
name: str
is_input: bool
data_type: Type[TypeVar(T)]
node: Optional[Node] = None
queue_size: int = 1000
full_strategy: QueueFullStrategy = 1
async initialize()[source]

Initialize the port’s queue.

connect_to(other)[source]

Create a connection from this port to another port.

Parameters:

other (Port[TypeVar(T)])

Return type:

Connection[TypeVar(T)]

async send(value)[source]

Send a value through this output port.

Parameters:

value (TypeVar(T))

Return type:

None

async receive(timeout=None)[source]

Receive a value from this input port.

Parameters:

timeout (Optional[float], default: None)

Return type:

Optional[TypeVar(T)]

has_data()[source]

Check if this input port has data available.

Return type:

bool

is_full()[source]

Check if this input port’s queue is full.

Return type:

bool

__init__(name, is_input, data_type, node=None, queue_size=1000, full_strategy=QueueFullStrategy.BLOCK)
Parameters:

Connection

class flow.Connection(source, target)[source]

Bases: Generic[T]

Represents a connection between two ports.

Parameters:
source: Port[TypeVar(T)]
target: Port[TypeVar(T)]
async transfer(value)[source]

Transfer a value through this connection.

Parameters:

value (TypeVar(T))

Return type:

None

__init__(source, target)
Parameters:

Executable Graph

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

Bases: object

Executable graph created directly by FlowBuilder.

Parameters:
__init__(name, nodes, connections, source_nodes)[source]
Parameters:
async run(duration=None, auto_stop=True)[source]

Execute the graph.

Parameters:
Return type:

None

async stop()[source]

Stop graph execution.

Return type:

None

Enumerations

Queue Full Strategy

class flow.QueueFullStrategy(value)[source]

Bases: Enum

Strategies for handling full queues.

Examples

Default strategy blocks when queue is full:

>>> QueueFullStrategy.BLOCK.name
'BLOCK'

Use in flow configuration:

>>> from flow import flow, QueueFullStrategy
>>> builder = (
...     flow()
...     .source([1, 2], int)
...     .transform(
...         lambda x: x * 2,
...         int,
...         queue_size=10,
...         full_strategy=QueueFullStrategy.DROP_OLD,
...     )
... )

Available strategies:

>>> [s.name for s in QueueFullStrategy]
['BLOCK', 'DROP_NEW', 'DROP_OLD', 'ERROR']
BLOCK = 1
DROP_NEW = 2
DROP_OLD = 3
ERROR = 4

Shutdown Mode

class flow.ShutdownMode(value)[source]

Bases: Enum

Different modes for shutting down the graph.

Examples

Available shutdown modes:

>>> [mode.name for mode in ShutdownMode]
['GRACEFUL', 'IMMEDIATE', 'FORCE']

Default is graceful shutdown:

>>> ShutdownMode.GRACEFUL.name
'GRACEFUL'
GRACEFUL = 1
IMMEDIATE = 2
FORCE = 3

Middleware System

Base Middleware

class flow.Middleware[source]

Bases: ABC

Base class for flow middleware.

Examples

Custom middleware implementation:

>>> class CounterMiddleware(Middleware):
...     def __init__(self):
...         self.count = 0
...
...     async def process(self, context, next_middleware):
...         self.count += 1
...         return await next_middleware(context)

>>> import asyncio
>>> counter = CounterMiddleware()
>>> async def handler(ctx):
...     return "done"
>>> result = asyncio.run(
...     counter.process(ProcessingContext("test", "transform"), handler)
... )
>>> counter.count
1
abstractmethod async process(context, next_middleware)[source]

Process the context and call next middleware in chain.

Parameters:
Return type:

Any

Processing Context

class flow.ProcessingContext(node_name, node_type, input_value=None, output_value=None, error=None, metadata=<factory>)[source]

Bases: object

Context passed to middleware with processing information.

Examples

Create a context:

>>> ctx = ProcessingContext(
...     node_name="transform_1", node_type="transform", input_value=42
... )
>>> ctx.node_name
'transform_1'
>>> ctx.metadata
{}

Add metadata:

>>> ctx.metadata["start_time"] = 123.45
>>> ctx.metadata["start_time"]
123.45
Parameters:
node_name: str
node_type: str
input_value: Any = None
output_value: Any = None
error: Optional[Exception] = None
metadata: Dict[str, Any]
__init__(node_name, node_type, input_value=None, output_value=None, error=None, metadata=<factory>)
Parameters:

Built-in Middleware

class flow.LoggingMiddleware(log_inputs=True, log_outputs=True)[source]

Bases: Middleware

Middleware that logs processing events.

Examples

Basic usage:

>>> logger = LoggingMiddleware(log_inputs=True, log_outputs=False)
>>> logger.log_inputs
True
>>> logger.log_outputs
False

In a flow (output would go to console):

>>> import asyncio
>>> from flow import flow
>>> async def example():
...     logger = LoggingMiddleware()
...     results = []
...     await (
...         flow()
...         .with_middleware(logger)
...         .source([1, 2], int)
...         .sink(results.append)
...         .execute(duration=0.5)
...     )
...     return results
>>> # asyncio.run(example())  # Would log to console
Parameters:

  • log_inputs (bool, default: True)

  • log_outputs (bool, default: True)

__init__(log_inputs=True, log_outputs=True)[source]
Parameters:

  • log_inputs (bool, default: True)

  • log_outputs (bool, default: True)

async process(context, next_middleware)[source]

Process the context and call next middleware in chain.

Parameters:
Return type:

Any

class flow.MetricsMiddleware[source]

Bases: Middleware

Middleware that collects processing metrics.

Examples

Track metrics across a flow:

>>> import asyncio
>>> from flow import flow
>>> async def example():
...     metrics = MetricsMiddleware()
...     results = []
...
...     await (
...         flow()
...         .with_middleware(metrics)
...         .source([1, 2, 3], int)
...         .filter(lambda x: x > 1)
...         .sink(results.append)
...         .execute(duration=0.5)
...     )
...
...     return metrics.get_metrics()
>>> stats = asyncio.run(example())
>>> stats["total_processed"] > 0  # Should have processed some items
True
>>> stats["total_errors"]
0
__init__()[source]
async process(context, next_middleware)[source]

Process the context and call next middleware in chain.

Parameters:
Return type:

Any

get_metrics()[source]

Get collected metrics.

Return type:

Dict[str, Any]

class flow.ThrottleMiddleware(delay_seconds)[source]

Bases: Middleware

Middleware that adds delays for rate limiting.

Examples

Create with 100ms delay:

>>> throttle = ThrottleMiddleware(0.1)
>>> throttle.delay_seconds
0.1

Use in a flow to limit processing rate:

>>> import asyncio
>>> import time
>>> from flow import flow
>>> async def example():
...     throttle = ThrottleMiddleware(0.05)  # 50ms delay
...     results = []
...
...     start = time.time()
...     await (
...         flow()
...         .source([1, 2], int)
...         .with_middleware(throttle)
...         .sink(results.append)
...         .execute(duration=1.0)
...     )
...     elapsed = time.time() - start
...
...     return len(results), elapsed > 0.1  # Should take > 100ms
>>> count, slow_enough = asyncio.run(example())
>>> count
2
>>> slow_enough
True
Parameters:

delay_seconds (float)

__init__(delay_seconds)[source]
Parameters:

delay_seconds (float)

async process(context, next_middleware)[source]

Process the context and call next middleware in chain.

Parameters:
Return type:

Any

class flow.RetryMiddleware(max_attempts=3, backoff=1.0)[source]

Bases: Middleware

Middleware that retries failed operations.

Examples

Create with custom settings:

>>> retry = RetryMiddleware(max_attempts=5, backoff=0.1)
>>> retry.max_attempts
5
>>> retry.backoff
0.1

Usage in a flow:

>>> import asyncio
>>> from flow import flow
>>> # The retry middleware can be used to handle transient failures
>>> retry = RetryMiddleware(max_attempts=3, backoff=0.1)
>>> # In practice, it would retry operations that fail temporarily
>>> # For example, network requests or database operations
Parameters:

  • max_attempts (int, default: 3)

  • backoff (float, default: 1.0)

__init__(max_attempts=3, backoff=1.0)[source]
Parameters:

  • max_attempts (int, default: 3)

  • backoff (float, default: 1.0)

async process(context, next_middleware)[source]

Process the context and call next middleware in chain.

Parameters:
Return type:

Any

Utility Functions

Async Utilities

flow.ensure_async(func)[source]

Ensure a function is async.

If the function is already async, returns it unchanged. If the function is sync, wraps it to be async.

Parameters:

func (Callable[..., Any]) – Function to ensure is async

Return type:

Callable[..., Awaitable[Any]]

Returns:

An async version of the function

Examples

Wrap a sync function:

>>> def sync_add(x, y):
...     return x + y
>>> async_add = ensure_async(sync_add)
>>> import asyncio
>>> asyncio.run(async_add(2, 3))
5

Already async function is unchanged:

>>> async def already_async(x):
...     return x * 2
>>> ensure_async(already_async) is already_async
True

Preserves function metadata:

>>> def documented_func(x):
...     '''This function doubles the input'''
...     return x * 2
>>> async_func = ensure_async(documented_func)
>>> async_func.__name__
'documented_func'
>>> async_func.__doc__
'This function doubles the input'

Async Iterator Factory

flow.AsyncIteratorAdapter(source)[source]

Create appropriate async iterator adapter based on source type.

This maintains backward compatibility with the old AsyncIteratorAdapter class.

Parameters:

source (Any)

Return type:

AsyncIterator[Any]