Quick Start

This guide will get you up and running with Flokkit Flow in just a few minutes.

Your First Flow

Let’s create a simple data processing pipeline:

from flow import flow
import asyncio

async def main():
    # Create a flow builder
    builder = flow("My First Pipeline")
    
    # Define processing functions
    def generate_numbers():
        """Generate numbers from 1 to 5"""
        for i in range(1, 6):
            yield i
    
    def square(x: int) -> int:
        """Square a number"""
        return x * x
    
    def print_result(x: int) -> None:
        """Print the result"""
        print(f"Result: {x}")
    
    # Build the pipeline
    (builder
        .source(generate_numbers)
        .to(square)
        .to(print_result))
    
    # Run for 2 seconds
    await builder.run(duration=2.0)

# Execute the flow
asyncio.run(main())

Output:

Result: 1
Result: 4
Result: 9
Result: 16
Result: 25

Understanding the Components

1. Flow Builder

The flow() function creates a builder for constructing your data pipeline:

builder = flow("Pipeline Name")

2. Source Nodes

Sources generate data for your pipeline:

def my_source():
    return "Hello, World!"

builder.source(my_source)

Sources can:

  • Return a single value

  • Yield multiple values (generator)

  • Be async functions

  • Read from files, databases, APIs, etc.

3. Transform Nodes

Transforms process data:

def uppercase(text: str) -> str:
    return text.upper()

builder.to(uppercase)

4. Sink Nodes

Sinks consume data (endpoints of your pipeline):

def save_to_file(data: str) -> None:
    with open("output.txt", "a") as f:
        f.write(data + "\n")

builder.to(save_to_file)

Working with Async Functions

Flokkit Flow seamlessly handles both sync and async functions:

async def fetch_data(url: str) -> dict:
    # Simulate API call
    await asyncio.sleep(0.1)
    return {"url": url, "status": "ok"}

async def process_async():
    builder = flow("Async Pipeline")
    
    urls = ["http://api1.com", "http://api2.com"]
    
    (builder
        .source(lambda: urls)
        .to(fetch_data)  # Async function works seamlessly
        .to(lambda data: print(f"Fetched: {data}")))
    
    await builder.run(duration=1.0)

Splitting and Merging

Process data in parallel paths:

async def parallel_processing():
    builder = flow("Parallel Pipeline")
    
    def generate_data():
        for i in range(10):
            yield i
    
    def process_even(x: int) -> str:
        return f"Even: {x}" if x % 2 == 0 else None
    
    def process_odd(x: int) -> str:
        return f"Odd: {x}" if x % 2 != 0 else None
    
    # Split into two paths
    source = builder.source(generate_data)
    
    # Process even numbers
    even_path = source.to(process_even).filter(lambda x: x is not None)
    
    # Process odd numbers  
    odd_path = source.to(process_odd).filter(lambda x: x is not None)
    
    # Merge results
    even_path.merge_with(odd_path).to(print)
    
    await builder.run(duration=2.0)

Error Handling

Flokkit Flow provides lifecycle hooks for proper error handling:

class RobustProcessor:
    async def on_start(self):
        """Initialize resources"""
        print("Starting processor...")
        self.resource = await self.connect_to_resource()
    
    async def on_stop(self):
        """Cleanup resources"""
        print("Stopping processor...")
        await self.resource.close()
    
    async def on_error(self, error: Exception):
        """Handle errors"""
        print(f"Error occurred: {error}")
        # Could implement retry logic here
    
    async def process(self, data):
        """Process data with error handling"""
        try:
            result = await self.resource.process(data)
            return result
        except Exception as e:
            await self.on_error(e)
            return None

Next Steps

Now that you understand the basics:

  1. Check out the Concepts page to understand Flokkit Flow’s architecture

  2. Explore Basic Usage for more patterns

  3. Learn about Backpressure for handling fast producers

  4. See Examples for real-world use cases

Common Patterns

ETL Pipeline

(builder
    .source(read_from_database)
    .to(transform_data)
    .to(validate_data)
    .to(load_to_warehouse))

Real-time Stream Processing

(builder
    .source(consume_kafka_stream)
    .to(parse_json)
    .filter(is_valid_event)
    .to(enrich_with_metadata)
    .to(publish_to_redis))

Batch Processing with Caching

(builder
    .source(list_files)
    .to(expensive_processing, cache=True)  # Cache results
    .to(save_results))