Node Reference

This section provides detailed documentation for all node types in Flokkit Flow.

Node Lifecycle

All nodes in Flokkit Flow follow a consistent lifecycle with optional hooks for resource management.

Lifecycle Methods

class MyNode(Node):
    async def on_start(self) -> None:
        """Called when the node starts.

        Use this to:
        - Initialize resources (database connections, file handles)
        - Set up internal state
        - Perform one-time setup operations
        """
        pass

    async def on_stop(self) -> None:
        """Called when the node stops.

        Use this to:
        - Close connections
        - Flush buffers
        - Clean up resources
        """
        pass

    async def on_error(self, error: Exception) -> None:
        """Called when an error occurs during processing.

        Use this to:
        - Log errors
        - Implement retry logic
        - Send alerts
        """
        pass

    async def on_complete(self) -> None:
        """Called when the node completes processing normally.

        Use this to:
        - Finalize results
        - Send completion notifications
        - Update metrics
        """
        pass

Creating Custom Nodes

Basic Custom Node

from flow import Node

class CustomProcessor(Node):
    def __init__(self, name: str, multiplier: int):
        super().__init__(name)
        self.multiplier = multiplier

    def _setup_ports(self) -> None:
        self.add_input_port("in", int)
        self.add_output_port("out", int)

    async def process(self) -> None:
        value = await self.get_input_port("in").receive()
        if value is not None:
            result = value * self.multiplier
            await self.get_output_port("out").send(result)

Stateful Node with Resources

class DatabaseWriter(Node):
    def __init__(self, name: str, connection_string: str):
        super().__init__(name)
        self.connection_string = connection_string
        self.connection = None
        self.batch = []
        self.batch_size = 100

    def _setup_ports(self) -> None:
        self.add_input_port("in", dict)

    async def on_start(self) -> None:
        self.connection = await asyncpg.connect(self.connection_string)

    async def on_stop(self) -> None:
        # Flush remaining batch
        if self.batch:
            await self._write_batch()

        # Close connection
        if self.connection:
            await self.connection.close()

    async def process(self) -> None:
        record = await self.get_input_port("in").receive()
        if record is not None:
            self.batch.append(record)

            if len(self.batch) >= self.batch_size:
                await self._write_batch()

    async def _write_batch(self) -> None:
        if not self.batch:
            return

        await self.connection.executemany(
            "INSERT INTO records (data) VALUES ($1)",
            [(json.dumps(record),) for record in self.batch]
        )
        self.batch.clear()

Port Management

Adding Ports

def _setup_ports(self) -> None:
    # Basic ports
    self.add_input_port("in", str)
    self.add_output_port("out", str)

    # With queue configuration
    self.add_input_port(
        "data",
        data_type=dict,
        queue_size=1000,
        full_strategy=QueueFullStrategy.DROP_OLD
    )

    # Multiple ports
    for i in range(3):
        self.add_output_port(f"out{i}", int)

Accessing Ports

async def process(self) -> None:
    # Get input
    in_port = self.get_input_port("in")
    value = await in_port.receive()

    # Check if port has data without blocking
    value = await in_port.receive(timeout=0)

    # Send output
    out_port = self.get_output_port("out")
    await out_port.send(result)

    # Send to specific port
    if condition:
        await self.get_output_port("out1").send(value)
    else:
        await self.get_output_port("out2").send(value)

Advanced Patterns

Multi-Input Processing

class Correlator(Node):
    """Correlates data from multiple sources."""

    def _setup_ports(self) -> None:
        self.add_input_port("primary", dict)
        self.add_input_port("secondary", dict)
        self.add_output_port("out", dict)

    async def process(self) -> None:
        # Wait for data from both inputs
        primary_task = asyncio.create_task(
            self.get_input_port("primary").receive()
        )
        secondary_task = asyncio.create_task(
            self.get_input_port("secondary").receive()
        )

        # Get first available
        done, pending = await asyncio.wait(
            {primary_task, secondary_task},
            return_when=asyncio.FIRST_COMPLETED
        )

        # Cancel pending
        for task in pending:
            task.cancel()

        # Process result
        for task in done:
            if task.result() is not None:
                await self.process_data(task.result())

Windowing Node

class WindowAggregator(Node):
    """Aggregates data in time windows."""

    def __init__(self, name: str, window_seconds: float):
        super().__init__(name)
        self.window_seconds = window_seconds
        self.window = []
        self.window_start = None

    def _setup_ports(self) -> None:
        self.add_input_port("in", float)
        self.add_output_port("out", dict)

    async def process(self) -> None:
        value = await self.get_input_port("in").receive(
            timeout=self.window_seconds
        )

        now = asyncio.get_event_loop().time()

        if self.window_start is None:
            self.window_start = now

        if value is not None:
            self.window.append(value)

        # Check if window expired
        if now - self.window_start >= self.window_seconds:
            await self._emit_window()

    async def _emit_window(self) -> None:
        if self.window:
            result = {
                "count": len(self.window),
                "sum": sum(self.window),
                "avg": sum(self.window) / len(self.window),
                "min": min(self.window),
                "max": max(self.window)
            }
            await self.get_output_port("out").send(result)

        self.window.clear()
        self.window_start = asyncio.get_event_loop().time()

Error Handling Node

class ResilientProcessor(Node):
    """Process with retry logic."""

    def __init__(self, name: str, max_retries: int = 3):
        super().__init__(name)
        self.max_retries = max_retries

    async def process(self) -> None:
        value = await self.get_input_port("in").receive()
        if value is None:
            return

        retries = 0
        while retries < self.max_retries:
            try:
                result = await self._process_with_errors(value)
                await self.get_output_port("out").send(result)
                break
            except Exception as e:
                retries += 1
                await self.on_error(e)
                if retries < self.max_retries:
                    await asyncio.sleep(2 ** retries)  # Exponential backoff
                else:
                    # Send to error output
                    await self.get_output_port("error").send({
                        "value": value,
                        "error": str(e),
                        "retries": retries
                    })

Best Practices

1. Resource Management: Always implement on_start() and on_stop() for resources

2. Error Handling: Use on_error() for logging and recovery

3. Buffering: Consider batching for I/O operations

4. Timeouts: Use receive timeouts to prevent indefinite blocking

5. Cancellation: Handle task cancellation gracefully

6. Type Safety: Always specify port data types

7. Testing: Design nodes to be testable in isolation