Welcome to Flokkit Flow
Flokkit Flow is a type-safe, async-first flow-based programming (FBP) framework for Python with zero dependencies.
Key Features
🔒 Type-safe - Full type safety across the entire flow graph
⚡ Async-first - Built on asyncio with automatic sync function adaptation
🚦 Backpressure - Bounded queues with configurable strategies
🔄 DAG Safety - Automatic cycle detection prevents deadlocks
📦 Zero Dependencies - Pure Python, no external packages required
🔌 Lifecycle Hooks - Proper resource management with init/cleanup
💾 Caching - Built-in memoization for expensive operations
🏭 Flexible Patterns - Support for both pipeline and server patterns
🎯 Middleware System - Extensible processing pipeline with logging, metrics, throttling, and retry support
Quick Example
from flow import flow
import asyncio
async def main():
# Build a simple pipeline
builder = flow("My Pipeline")
def generate_numbers():
return 42
def double(x):
return x * 2
def print_result(x):
print(f"Result: {x}")
# Connect the nodes
(builder
.source(generate_numbers, int)
.transform(double, int)
.to(print_result))
# Run the pipeline
await builder.execute(duration=1.0)
asyncio.run(main())
Table of Contents
Getting Started
User Guide
API Reference
Examples