pipetee 0.3.0

A pipeline framework for processing data

Pipeline-Tee

A powerful and flexible Python pipeline processing framework with advanced flow control, parallel execution, visualization, and async support.

Features

• 🔄 Flexible Pipeline Processing: Build complex data processing pipelines with branching and conditional execution
• ⚡ Async Support: Built with asyncio for efficient asynchronous processing
• 🚀 Parallel Execution: True concurrent processing with 1-to-N-to-1 patterns (Fan Out/Fan In)
  • Selective Parallelism: Choose which stages run in parallel vs sequential
  • Multiple Parallel Groups: Mix parallel and sequential execution in the same pipeline
  • Performance Gains: 30-50% faster execution for independent tasks
  • Concurrent I/O: Ideal for API calls, database operations, file processing
• 📊 Visualization: Rich pipeline visualization with graphviz and matplotlib
  • Generate pipeline structure diagrams showing stages and connections
  • Create execution timeline plots with timing information
  • Track stage status with color-coded nodes and status icons
  • Parallel execution visualization with overlapping timelines
  • Export as PNG, SVG, or PDF files
• 🔀 Advanced Flow Control: Support for branching, skipping, jumping between stages
• 📝 Comprehensive Logging: Detailed logging with configurable levels
• 🔍 State Tracking: Pipeline state management and execution metrics
• 🎯 Post-Processing: Add post-processors to modify stage outputs
• ⚠️ Error Handling: Robust error handling and propagation

Requirements

• Python 3.8+
• Optional visualization dependencies:
  • graphviz (for pipeline structure diagrams)
  • matplotlib (for execution timeline plots)

Project Structure

pipetee/
├── src/
│   └── pipetee/
│       ├── __init__.py
│       ├── pipeline.py          # Core pipeline implementation
│       ├── models/
│       │   ├── __init__.py
│       │   ├── stage.py        # Stage models and decisions
│       │   └── visual.py       # Visualization models
│       └── utils/
│           ├── logging_config.py
│           └── visualization.py # Visualization utilities
├── tests/
│   ├── test_pipeline.py
│   ├── test_visual.py
│   └── test_models.py
├── examples/
│   ├── parallel_pipeline_demo.py      # NEW: 1-to-N-to-1 parallel pattern
│   ├── mixed_parallel_demo.py         # NEW: Mixed parallel/sequential
│   ├── complex_pipeline_demo.py
│   └── complex_branching_pipeline.py
└── pyproject.toml

Installation

You can install Pipeline-Tee directly from PyPI:

# Install core package
pip install pipetee

# Install with visualization dependencies
pip install pipetee[viz]

For development installation:

# Clone the repository
git clone https://github.com/yudataguy/pipeline-tee.git
cd pipeline-tee

# Install in development mode
pip install -e .

# Install with visualization dependencies
pip install -e ".[viz]"

Basic Usage

Sequential Pipeline (Traditional)

from pipetee.pipeline import Pipeline, PipelineStage, StageResult
from pipetee.models.stage import StageDecision
from pipetee.utils.visualization import save_pipeline_visualization

# Define a custom pipeline stage
class DataProcessingStage(PipelineStage[str, str]):
    async def process(self, data: str) -> StageResult[str]:
        processed_data = data.upper()
        return StageResult(
            success=True,
            data=processed_data,
            decision=StageDecision.CONTINUE
        )

# Create and configure pipeline
pipeline = Pipeline()
pipeline.add_stage("process_data", DataProcessingStage())

# Process data
result = await pipeline.process("hello world")

# Save visualization diagrams
structure_path, timeline_path = save_pipeline_visualization(
    pipeline,
    output_dir="pipeline_viz",
    base_name="my_pipeline",
    format="png"  # or "svg" or "pdf"
)

print(f"Pipeline structure saved to: {structure_path}")
print(f"Execution timeline saved to: {timeline_path}")

Parallel Pipeline (NEW!) ⚡

import asyncio
from pipetee.pipeline import Pipeline, PipelineStage, StageResult
from pipetee.models.stage import StageDecision

# Stage that triggers parallel execution
class DataPreparationStage(PipelineStage):
    async def process(self, data):
        # Prepare data for parallel processing
        prepared_data = {"input": data, "timestamp": "2024-01-01"}

        # Trigger parallel execution of multiple stages
        return StageResult(
            success=True,
            data=prepared_data,
            decision=StageDecision.PARALLEL_TO,
            next_stage="text_analyzer,image_processor,data_validator"  # Run in parallel
        )

# Parallel stage 1: Text Analysis
class TextAnalyzerStage(PipelineStage):
    async def process(self, data):
        await asyncio.sleep(1.0)  # Simulate text processing
        return StageResult(success=True, data={"stage": "text_analyzer", "result": "analyzed"})

# Parallel stage 2: Image Processing
class ImageProcessorStage(PipelineStage):
    async def process(self, data):
        await asyncio.sleep(1.5)  # Simulate image processing
        return StageResult(success=True, data={"stage": "image_processor", "result": "processed"})

# Parallel stage 3: Data Validation
class DataValidatorStage(PipelineStage):
    async def process(self, data):
        await asyncio.sleep(0.8)  # Simulate validation
        return StageResult(success=True, data={"stage": "data_validator", "result": "valid"})

# Aggregation stage (N-to-1)
class ResultAggregatorStage(PipelineStage):
    async def process(self, parallel_results):
        # parallel_results is a list of results from all parallel stages
        aggregated = {"parallel_results": parallel_results, "status": "aggregated"}
        return StageResult(success=True, data=aggregated)

# Create pipeline with parallel execution
pipeline = Pipeline()
pipeline.add_stage("data_preparation", DataPreparationStage())
pipeline.add_stage("text_analyzer", TextAnalyzerStage())        # Parallel group
pipeline.add_stage("image_processor", ImageProcessorStage())    # Parallel group
pipeline.add_stage("data_validator", DataValidatorStage())      # Parallel group
pipeline.add_stage("result_aggregator", ResultAggregatorStage()) # Sequential

# Execute pipeline
result = await pipeline.process({"data": "sample input"})

# Pipeline flow:
# data_preparation → [text_analyzer + image_processor + data_validator] → result_aggregator
#                    ↑ These 3 stages run concurrently ↑

Parallel Execution Patterns

1-to-N-to-1 (Fan Out/Fan In)

The most common parallel pattern where one stage distributes work to multiple parallel stages, then aggregates the results:

# Sequential Stage → [Parallel Stage A + Parallel Stage B + Parallel Stage C] → Aggregation Stage

Benefits:

• Concurrent I/O: Database calls, API requests, file operations run simultaneously
• Independent Processing: Text analysis + image processing + data validation
• Performance: 30-50% faster execution for suitable workloads

Selective Parallelism

Not all stages need to be parallel. Mix and match based on your needs:

# Pipeline with mixed execution:
# Init → Analysis → [Parallel Group 1] → Aggregation → Middle → [Parallel Group 2] → Final

Use Parallel When:

• ✅ Tasks are independent and can run concurrently
• ✅ I/O-bound operations (API calls, database queries)
• ✅ CPU-intensive computations that can be parallelized
• ✅ Different transformations on the same input data

Stay Sequential When:

• ⚠️ Stage B depends on Stage A's output
• ⚠️ Shared resource modifications (database updates)
• ⚠️ Setup/teardown operations
• ⚠️ Very fast operations (<100ms) where parallel overhead exceeds benefits

Visualization Examples

The pipeline visualization utilities can generate two types of diagrams:

1. Pipeline Structure (using graphviz):

   • Shows stages and their connections
   • Color-coded nodes based on stage status
   • Status icons (🔄 pending, ⚡ running, ✅ completed, ⏭️ skipped, ❌ failed)
   • Default sequence and conditional branching paths
   • NEW: Parallel execution groups clearly marked

2. Execution Timeline (using matplotlib):

   • Shows execution flow over time
   • Color-coded bars for stage duration
   • Start and end times for each stage
   • Status icons and stage names
   • Time-based x-axis
   • NEW: Overlapping bars show parallel execution

To generate visualizations, use the save_pipeline_visualization function from pipetee.utils.visualization. Make sure to install the visualization dependencies with pip install -e ".[viz]".

Advanced Features

Branching and Conditions

from pipetee.pipeline import Condition

# Add branching condition
condition = Condition("is_valid", lambda x: len(x) > 5)
stage.add_branch_condition(condition, "validation_stage")

# Add skip condition
stage.add_skip_condition(Condition("skip_empty", lambda x: not x))

Parallel Branch Conditions (NEW!)

# Add parallel branching condition
parallel_condition = Condition("high_volume", lambda x: x.get("volume", 0) > 1000)
stage.add_parallel_branch_condition(parallel_condition, ["heavy_processor_1", "heavy_processor_2", "heavy_processor_3"])

Post-Processing

# Add post-processor to a stage
stage.add_post_processor(lambda x: x.strip())

Performance Monitoring

# The pipeline automatically tracks execution metrics
result = await pipeline.process(data)

if result.metadata:
    execution_path = result.metadata.get('execution_path', [])
    duration = result.metadata.get('pipeline_end') - result.metadata.get('pipeline_start')
    print(f"Executed stages: {execution_path}")
    print(f"Total time: {duration.total_seconds():.2f} seconds")

Examples

Check out the examples/ directory for comprehensive demonstrations:

• parallel_pipeline_demo.py: 1-to-N-to-1 parallel execution with performance analysis
• mixed_parallel_demo.py: Mixed parallel and sequential execution patterns
• visualization_demo.py: Pipeline visualization features
• complex_pipeline_demo.py: Complex data processing workflows
• complex_branching_pipeline.py: Advanced flow control features

Run any example:

# Parallel execution demos
python examples/parallel_pipeline_demo.py
python examples/mixed_parallel_demo.py

# Traditional demos
python examples/visualization_demo.py
python examples/complex_pipeline_demo.py

Development

1. Clone the repository
2. Install development dependencies:

   python -m pip install -e ".[dev,viz]"
   

3. Run tests:

   pytest
   

Performance Benchmarks

Example performance improvements with parallel execution:

Traditional Sequential Pipeline: 4.9 seconds
├── Data Processing: 1.5s
├── Text Analysis: 1.0s
├── Image Processing: 1.2s
└── Validation: 0.8s

Parallel Pipeline: 3.1 seconds (37% faster)
├── Data Processing: 1.5s
└── [Text + Image + Validation]: 1.2s (concurrent)

Real-world scenarios where parallel execution excels:

• API data aggregation from multiple sources
• Batch processing of files with different operations
• ML pipeline with parallel feature extraction
• Data validation + transformation + enrichment workflows

MIT License