flux-pipeline 0.1.1

A simple framework for building composable, asynchronous workflows

Flux - Tiny Pipeline Framework

A minimal, composable pipeline framework for building async workflows with persistence, parallel execution, and pause/resume capabilities.

🚀 Quick Start

Installation

# Create and activate virtual environment
python3 -m venv venv
source venv/bin/activate

# Install dependencies
pip install -r requirements.txt

# Install package in development mode
pip install -e .

Basic Example

import asyncio
from flux import Node

@Node
def step1(input):
    return f"processed: {input}"

@Node
def step2(input):
    return f"{input} -> complete"

# Chain nodes with pipe operator
pipeline = step1 | step2

# Execute
result = await pipeline("hello")
# Output: "processed: hello -> complete"

📁 Core Concepts

1. Nodes - Building Blocks

Decorate any function to make it a pipeline node:

@Node
def my_function(input):
    return input.upper()

Nodes can be sync or async, and automatically handle persistence and shared state.

2. Pipelines - Composition with |

Chain nodes sequentially using the pipe operator:

pipeline = node1 | node2 | node3
result = await pipeline(input_data)

3. Parallel Execution

Run multiple nodes in parallel:

from flux import Parallel

parallel = Parallel(node1, node2, node3)
results = await parallel(input)  # Returns list of results

4. Conditional Branching

Route execution based on conditions:

from flux import Conditional

@Node
def check_type(input):
    return "number" if isinstance(input, int) else "text"

pipeline = Conditional(
    check_type,
    {
        "number": number_handler,
        "text": text_handler,
        "": default_handler  # Default branch
    }
)

5. Persistence & State

Track execution state and share data between nodes:

@Node
def save_data(input, persistence=None):
    if persistence:
        persistence.save("key", "value")
    return input

@Node
def load_data(input, persistence=None):
    if persistence:
        value = persistence.get("key")
    return input

6. Pause/Resume

Pause and resume pipeline execution:

@Node
def pausable_step(input, persistence=None, node=None):
    if should_pause:
        node.pause(persistence)
    return input

# Execution pauses, can be resumed later

🛠️ Framework Features

Node Capabilities

• Auto-detection: Automatically detects sync/async functions
• Dependency injection: Supports shared, persistence, storage, observability
• Composable: Chain with | operator or use in Parallel/Conditional

Persistence API

persistence.save(key, value)      # Save state
persistence.get(key, default)     # Get state
persistence.delete(key)           # Delete state
persistence.get_history()         # Get execution history
persistence.set_status(status)    # Mark run status
persistence.pause()               # Pause execution

Run Management

Persistence.list_runs()           # List all tracked runs
Persistence.get_run(run_id)       # Get specific run data
Persistence.cleanup_run(run_id)   # Clean up run data

📝 Complete Example

import asyncio
from flux import Node, Parallel

@Node
def fetch_data(input):
    return f"data-{input}"

@Node
def process_data(input, persistence=None):
    if persistence:
        persistence.save("processed", input)
    return input.upper()

@Node
def transform(input):
    return f"transformed-{input}"

@Node
def combine(inputs: list):
    return " + ".join(inputs)

# Build pipeline: parallel processing then combine
parallel = Parallel(
    fetch_data | process_data,
    transform
)
pipeline = parallel | combine

# Execute
result = await pipeline("start")
# Output: "DATA-DATA-START + transformed-start"

🔧 Advanced Features

Custom Storage Backend

class CustomStorage:
    def save(self, key: str, value: Any) -> None: ...
    def get(self, key: str) -> Any: ...
    def delete(self, key: str) -> bool: ...

pipeline = my_pipeline(input, storage=CustomStorage())

Observability

def logger(action: str, key: str, value: Any) -> None:
    print(f"{action}: {key} = {value}")

pipeline = my_pipeline(input, observability=logger)

Execution History

result = await pipeline(input)
history = persistence.get_history()

for entry in history:
    print(f"{entry['node_name']}: {entry['duration']:.4f}s")

📁 Project Structure

flux/
├── __init__.py          # Tiny framework (~400 lines)
├── requirements.txt     # Dependencies
├── setup.py            # Package setup
└── tests/              # Test suite

🎯 Design Philosophy

• Tiny: ~400 lines of core framework code
• Composable: Build complex workflows from simple functions
• Flexible: Sync/async, parallel, conditional execution
• Observable: Built-in persistence and execution tracking
• Zero Dependencies: Only Python standard library (asyncio)

🧪 Testing

# Run tests
python -m pytest

# Run quick test
python __init__.py

📄 License

This project is licensed under the MIT License.