StreamForge

A Python library for building flexible data transformation pipelines with real-time processing capabilities. Create composable data processing workflows that handle both batch and streaming data with configurable transformations and error handling.

Installation

pip install streamforge

For development:

git clone https://github.com/yourusername/streamforge.git
cd streamforge
pip install -e .

Quick Start

Basic Pipeline

from streamforge import Pipeline, Transform

# Create a simple transformation pipeline
pipeline = Pipeline() \
    .add(Transform.map(lambda x: x * 2)) \
    .add(Transform.filter(lambda x: x > 10)) \
    .add(Transform.map(str))

# Process batch data
result = pipeline.process([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
print(result)  # ['12', '14', '16', '18', '20']

Async Processing

import asyncio
from streamforge import AsyncPipeline, AsyncTransform

async def async_double(x):
    await asyncio.sleep(0.1)  # Simulate async work
    return x * 2

async def main():
    pipeline = AsyncPipeline() \
        .add(AsyncTransform.map(async_double)) \
        .add(AsyncTransform.filter(lambda x: x > 5))
    
    result = await pipeline.process([1, 2, 3, 4, 5])
    print(result)  # [6, 8, 10]

asyncio.run(main())

Error Handling

from streamforge import Pipeline, Transform, RetryConfig

pipeline = Pipeline(retry_config=RetryConfig(max_retries=3, delay=1.0)) \
    .add(Transform.map(lambda x: 1 / x))  # May raise ZeroDivisionError

# Errors are logged and problematic items are filtered out
result = pipeline.process([1, 2, 0, 3, 4])  # [1.0, 0.5, 0.333..., 0.25]

Features

• Chainable Operations: Build pipelines with method chaining
• Async Support: Handle both sync and async transformations
• Error Recovery: Built-in retry mechanisms and error handling
• Flexible Processing: Support for batch and streaming data
• Type Safety: Full type hints for better development experience

License

MIT License