Core Concepts

Understanding Flokkit Flow’s core concepts will help you build efficient and maintainable data pipelines.

Flow-Based Programming (FBP)

Flokkit Flow implements the Flow-Based Programming paradigm, where:

  • Applications are networks of “black box” processes

  • Processes communicate via message passing

  • Connections are defined separately from processes

  • Data flows through the network as discrete packets

Benefits of FBP

  1. Modularity: Each component has a single responsibility

  2. Reusability: Components can be reused in different flows

  3. Testability: Components can be tested in isolation

  4. Visual Design: Flows can be visualized as directed graphs

  5. Natural Parallelism: Components run concurrently by default

Nodes

Nodes are the building blocks of a flow. Flokkit Flow provides several node types:

Source Nodes

Generate data for the pipeline:

def file_reader():
    with open("data.txt") as f:
        for line in f:
            yield line.strip()

Transform Nodes

Process data and produce outputs:

def parse_json(line: str) -> dict:
    return json.loads(line)

Sink Nodes

Consume data without producing outputs:

def write_to_database(record: dict) -> None:
    db.insert(record)

Special Nodes

  • Filter: Conditionally pass data through

  • Tap: Observe data without consuming it

  • Split: Duplicate data to multiple paths

  • Merge: Combine multiple streams

Ports and Connections

Ports

Every node has input and/or output ports:

  • Input Port: Receives data from upstream nodes

  • Output Port: Sends data to downstream nodes

Ports are typed, ensuring type safety across connections.

Connections

Connections link output ports to input ports:

# Connects output of 'source' to input of 'processor'
builder.graph.connect(
    source.name, "out",     # Source node, output port
    processor.name, "in"    # Target node, input port
)

Queues and Backpressure

Bounded Queues

Each connection uses a bounded asyncio.Queue to:

  1. Prevent memory exhaustion

  2. Provide natural backpressure

  3. Enable concurrent execution

Queue Strategies

When a queue is full, Flokkit Flow supports different strategies:

from flow import QueueFullStrategy

# Block until space available (default)
QueueFullStrategy.BLOCK

# Drop new items
QueueFullStrategy.DROP_NEW  

# Drop oldest items
QueueFullStrategy.DROP_OLD

# Raise an error
QueueFullStrategy.ERROR

Example:

# Configure queue size and strategy
(builder
    .source(fast_producer)
    .to(slow_consumer, 
        queue_size=100,
        full_strategy=QueueFullStrategy.DROP_OLD))

Type Safety

Flokkit Flow leverages Python’s type system for safety:

from typing import List

def parse_numbers(text: str) -> List[int]:
    return [int(x) for x in text.split(",")]

def sum_numbers(numbers: List[int]) -> int:
    return sum(numbers)

# This will type-check correctly
(builder
    .source(lambda: "1,2,3,4,5")
    .to(parse_numbers)  # str -> List[int]
    .to(sum_numbers)    # List[int] -> int
    .to(print))         # int -> None

Async Execution

Automatic Adaptation

Flokkit Flow automatically adapts sync functions to async:

def sync_function(x: int) -> int:
    time.sleep(0.1)  # Blocking call
    return x * 2

# Flokkit Flow wraps this in asyncio.to_thread()
builder.to(sync_function, is_blocking=True)

Native Async Support

Async functions run directly on the event loop:

async def fetch_data(url: str) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()

Graph Execution

DAG Verification

Flokkit Flow verifies that your graph is a Directed Acyclic Graph (DAG):

# This will raise an error - cycles aren't allowed
A -> B -> C -> A  # ❌ Cycle detected!

Execution Model

  1. Initialization: All nodes start their process() coroutines

  2. Data Flow: Data flows through connections via queues

  3. Concurrency: Nodes process data concurrently

  4. Completion: Nodes complete when sources are exhausted

  5. Shutdown: Resources are cleaned up via lifecycle hooks

Lifecycle Management

Nodes can implement lifecycle methods:

class StatefulNode:
    async def on_start(self):
        """Called before processing starts"""
        self.connection = await database.connect()
    
    async def on_stop(self):
        """Called during shutdown"""
        await self.connection.close()
    
    async def on_error(self, error: Exception):
        """Called when an error occurs"""
        logger.error(f"Error in node: {error}")
    
    async def on_complete(self):
        """Called when processing completes normally"""
        await self.flush_buffers()

Pipeline vs Server Patterns

Pipeline Pattern (auto_stop=True)

For batch processing that completes:

builder = flow("ETL Pipeline", auto_stop=True)
# Automatically stops when sources are exhausted

Server Pattern (auto_stop=False)

For continuous processing:

builder = flow("API Server", auto_stop=False)
# Runs until manually stopped

Caching and Memoization

Enable caching for expensive operations:

@cache  # or builder.to(expensive_op, cache=True)
def expensive_operation(data: str) -> Result:
    # This will only run once per unique input
    return complex_calculation(data)

Best Practices

  1. Keep nodes simple: Each node should do one thing well

  2. Use type hints: Enable type checking for safety

  3. Handle errors: Implement lifecycle hooks for robustness

  4. Configure queues: Set appropriate sizes for your workload

  5. Monitor performance: Use tap nodes for observability

  6. Test components: Nodes should be testable in isolation

Next Steps