Middleware System

Flokkit Flow’s middleware system provides a powerful way to add cross-cutting concerns to your data processing pipelines without modifying the core logic. Middleware can intercept and modify the processing of data at each node in your flow.

Overview

Middleware in Flokkit Flow follows the chain-of-responsibility pattern, where each middleware can:

Inspect or modify input data before processing
Inspect or modify output data after processing
Handle errors during processing
Add metadata or context to the processing pipeline
Implement concerns like logging, metrics, rate limiting, and retry logic

Basic Usage

from flow import flow, LoggingMiddleware

# Create middleware instance
logger = LoggingMiddleware()

# Apply to a flow
builder = flow("My Pipeline")
await (
    builder
    .with_middleware(logger)  # Apply to all subsequent nodes
    .source(my_source, int)
    .transform(my_transform, str)
    .to(my_sink)
    .execute()
)

Built-in Middleware

LoggingMiddleware

Logs data as it flows through each node.

from flow import LoggingMiddleware

# Log both inputs and outputs
logger = LoggingMiddleware(log_inputs=True, log_outputs=True)

# Log only outputs
output_logger = LoggingMiddleware(log_inputs=False, log_outputs=True)

Output Example:

[transform_1] Input: 42
[transform_1] Output: 84
[filter_1] Input: 84

MetricsMiddleware

Collects processing metrics for monitoring and analysis.

from flow import MetricsMiddleware

metrics = MetricsMiddleware()

# ... run your pipeline ...

# Get collected metrics
stats = metrics.get_metrics()
print(f"Total processed: {stats['total_processed']}")
print(f"Total errors: {stats['total_errors']}")
print(f"By node: {stats['by_node']}")

Metrics Structure:

{
    'total_processed': 150,
    'total_errors': 2,
    'by_node': {
        'source_1': {'processed': 50, 'errors': 0},
        'transform_1': {'processed': 50, 'errors': 1},
        'sink_1': {'processed': 50, 'errors': 1}
    }
}

ThrottleMiddleware

Adds delays between processing to control throughput.

from flow import ThrottleMiddleware

# Add 100ms delay after each processing
throttle = ThrottleMiddleware(delay_seconds=0.1)

# Useful for:
# - Rate limiting API calls
# - Preventing overwhelming downstream systems
# - Testing backpressure behavior

RetryMiddleware

Automatically retries failed operations with exponential backoff.

from flow import RetryMiddleware

# Retry up to 3 times with exponential backoff
retry = RetryMiddleware(max_attempts=3, backoff=1.0)

# Customize retry behavior
aggressive_retry = RetryMiddleware(
    max_attempts=5,
    backoff=0.5  # Start with 0.5s, then 1s, 2s, 4s, 8s
)

Creating Custom Middleware

To create custom middleware, inherit from the Middleware base class:

from flow import Middleware, ProcessingContext
from typing import Any, Awaitable, Callable

class TimingMiddleware(Middleware):
    """Measure processing time for each node."""
    
    def __init__(self):
        self.timings = {}
    
    async def process(
        self,
        context: ProcessingContext,
        next_middleware: Callable[[ProcessingContext], Awaitable[Any]]
    ) -> Any:
        import time
        
        start = time.time()
        try:
            # Call the next middleware/handler
            result = await next_middleware(context)
            
            # Record timing
            elapsed = time.time() - start
            if context.node_name not in self.timings:
                self.timings[context.node_name] = []
            self.timings[context.node_name].append(elapsed)
            
            return result
        except Exception as e:
            # Still record timing for failed operations
            elapsed = time.time() - start
            print(f"[{context.node_name}] Failed after {elapsed:.3f}s: {e}")
            raise

ProcessingContext

The ProcessingContext provides information about the current processing:

@dataclass
class ProcessingContext:
    node_name: str          # Name of the current node
    node_type: str          # Type of node (source, transform, filter, etc.)
    input_value: Any = None # Input value being processed
    output_value: Any = None # Output value after processing
    error: Optional[Exception] = None  # Any error that occurred
    metadata: Dict[str, Any] = field(default_factory=dict)  # Custom metadata

Middleware Composition

Multiple middleware can be combined:

from flow import flow, LoggingMiddleware, MetricsMiddleware, RetryMiddleware

# Create middleware instances
logger = LoggingMiddleware()
metrics = MetricsMiddleware()
retry = RetryMiddleware(max_attempts=3)

# Apply multiple middleware
builder = flow("Robust Pipeline")
await (
    builder
    .with_middleware(logger, metrics, retry)  # Applied in order
    .source(unreliable_source, dict)
    .transform(process_data, dict)
    .to(save_result)
    .execute()
)

Middleware are executed in the order they are added:

First middleware sees the original input
Each middleware can modify the context
Last middleware is closest to the actual processing

Advanced Patterns

Conditional Middleware

class ConditionalMiddleware(Middleware):
    """Only apply middleware logic based on conditions."""
    
    def __init__(self, condition_fn: Callable[[ProcessingContext], bool]):
        self.condition_fn = condition_fn
    
    async def process(self, context, next_middleware):
        if self.condition_fn(context):
            # Apply special logic
            print(f"Condition met for {context.node_name}")
        
        return await next_middleware(context)

# Only log transform nodes
log_transforms = ConditionalMiddleware(
    lambda ctx: ctx.node_type == "transform"
)

Error Handling Middleware

class ErrorHandlerMiddleware(Middleware):
    """Handle specific errors gracefully."""
    
    def __init__(self, error_handlers: Dict[Type[Exception], Callable]):
        self.error_handlers = error_handlers
    
    async def process(self, context, next_middleware):
        try:
            return await next_middleware(context)
        except Exception as e:
            handler = self.error_handlers.get(type(e))
            if handler:
                return handler(e, context)
            raise

# Usage
error_handler = ErrorHandlerMiddleware({
    ValueError: lambda e, ctx: None,  # Skip bad values
    KeyError: lambda e, ctx: {},      # Return empty dict for missing keys
})

State-Tracking Middleware

class StateMiddleware(Middleware):
    """Track state across processing."""
    
    def __init__(self):
        self.state = {}
        self.processed_count = 0
    
    async def process(self, context, next_middleware):
        self.processed_count += 1
        
        # Add state to context
        context.metadata['global_count'] = self.processed_count
        context.metadata['state'] = self.state
        
        result = await next_middleware(context)
        
        # Update state based on result
        if result and isinstance(result, dict):
            self.state.update(result)
        
        return result

Best Practices

Keep Middleware Focused: Each middleware should have a single responsibility
Avoid Heavy Processing: Middleware should be lightweight to avoid impacting performance
Handle Errors Gracefully: Always consider error cases in your middleware
Use Context Metadata: Store middleware-specific data in context.metadata
Make Middleware Reusable: Design middleware to be configurable and reusable across flows

Performance Considerations

Middleware adds overhead to each processing step
Use throttling middleware carefully as it directly impacts throughput
Metrics collection has minimal overhead but can accumulate memory for long-running flows
Retry middleware can significantly increase processing time for unreliable operations

Examples

See the examples/ directory for complete middleware examples:

13_middleware_basics.py - Introduction to middleware
14_custom_middleware.py - Creating custom middleware
15_middleware_composition.py - Combining multiple middleware