Middleware API

Middleware provides a powerful way to add cross-cutting concerns to your flows.

Core Classes

Middleware Base Class

class flow.Middleware[source]

Bases: ABC

Base class for flow middleware.

Examples

Custom middleware implementation:

>>> class CounterMiddleware(Middleware):
...     def __init__(self):
...         self.count = 0
...
...     async def process(self, context, next_middleware):
...         self.count += 1
...         return await next_middleware(context)

>>> import asyncio
>>> counter = CounterMiddleware()
>>> async def handler(ctx):
...     return "done"
>>> result = asyncio.run(
...     counter.process(ProcessingContext("test", "transform"), handler)
... )
>>> counter.count
1

abstractmethod async process(context, next_middleware)[source]

Process the context and call next middleware in chain.

Parameters:

context (ProcessingContext)
next_middleware (Callable[[ProcessingContext], Awaitable[Any]])

Return type:

Any

All middleware must implement the process method:

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

class MyMiddleware(Middleware):
    async def process(
        self,
        context: ProcessingContext,
        next_middleware: Callable[[ProcessingContext], Awaitable[Any]]
    ) -> Any:
        # Pre-processing
        print(f"Before {context.node_name}")

        # Call next middleware or node
        result = await next_middleware(context)

        # Post-processing
        print(f"After {context.node_name}")

        return result

ProcessingContext

class flow.ProcessingContext(node_name, node_type, input_value=None, output_value=None, error=None, metadata=<factory>)[source]

Bases: object

Context passed to middleware with processing information.

Examples

Create a context:

>>> ctx = ProcessingContext(
...     node_name="transform_1", node_type="transform", input_value=42
... )
>>> ctx.node_name
'transform_1'
>>> ctx.metadata
{}

Add metadata:

>>> ctx.metadata["start_time"] = 123.45
>>> ctx.metadata["start_time"]
123.45

Parameters:

node_name (str)
node_type (str)
input_value (Any, default: None)
output_value (Any, default: None)
error (Optional[Exception], default: None)
metadata (Dict[str, Any], default: <factory>)

node_name: str

node_type: str

input_value: Any = None

output_value: Any = None

error: Optional[Exception] = None

metadata: Dict[str, Any]

__init__(node_name, node_type, input_value=None, output_value=None, error=None, metadata=<factory>)

Parameters:

node_name (str)
node_type (str)
input_value (Any, default: None)
output_value (Any, default: None)
error (Optional[Exception], default: None)
metadata (Dict[str, Any], default: <factory>)

The context provides information about the current processing:

async def process(self, context, next_middleware):
    # Access context properties
    print(f"Node: {context.node_name}")
    print(f"Type: {context.node_type}")
    print(f"Input: {context.input_value}")

    result = await next_middleware(context)

    print(f"Output: {context.output_value}")

    # Use metadata for passing data between middleware
    context.metadata['processed_by'] = self.__class__.__name__

    return result

Built-in Middleware

LoggingMiddleware

class flow.LoggingMiddleware(log_inputs=True, log_outputs=True)[source]

Bases: Middleware

Middleware that logs processing events.

Examples

Basic usage:

>>> logger = LoggingMiddleware(log_inputs=True, log_outputs=False)
>>> logger.log_inputs
True
>>> logger.log_outputs
False

In a flow (output would go to console):

>>> import asyncio
>>> from flow import flow
>>> async def example():
...     logger = LoggingMiddleware()
...     results = []
...     await (
...         flow()
...         .with_middleware(logger)
...         .source([1, 2], int)
...         .sink(results.append)
...         .execute(duration=0.5)
...     )
...     return results
>>> # asyncio.run(example())  # Would log to console

Parameters:

log_inputs (bool, default: True)
log_outputs (bool, default: True)

__init__(log_inputs=True, log_outputs=True)[source]

Parameters:

log_inputs (bool, default: True)
log_outputs (bool, default: True)

async process(context, next_middleware)[source]

Process the context and call next middleware in chain.

Parameters:

context (ProcessingContext)
next_middleware (Callable[[ProcessingContext], Awaitable[Any]])

Return type:

Any

Example usage:

from flow import flow, LoggingMiddleware

# Basic logging
logger = LoggingMiddleware()

# Detailed logging
logger = LoggingMiddleware(
    log_inputs=True,
    log_outputs=True,
    log_errors=True
)

builder.with_middleware(logger)

MetricsMiddleware

class flow.MetricsMiddleware[source]

Bases: Middleware

Middleware that collects processing metrics.

Examples

Track metrics across a flow:

>>> import asyncio
>>> from flow import flow
>>> async def example():
...     metrics = MetricsMiddleware()
...     results = []
...
...     await (
...         flow()
...         .with_middleware(metrics)
...         .source([1, 2, 3], int)
...         .filter(lambda x: x > 1)
...         .sink(results.append)
...         .execute(duration=0.5)
...     )
...
...     return metrics.get_metrics()
>>> stats = asyncio.run(example())
>>> stats["total_processed"] > 0  # Should have processed some items
True
>>> stats["total_errors"]
0

__init__()[source]

async process(context, next_middleware)[source]

Process the context and call next middleware in chain.

Parameters:

context (ProcessingContext)
next_middleware (Callable[[ProcessingContext], Awaitable[Any]])

Return type:

Any

get_metrics()[source]

Get collected metrics.

Return type:: Dict[str, Any]

Example usage:

from flow import flow, MetricsMiddleware

metrics = MetricsMiddleware()

await builder.with_middleware(metrics).execute()

# Get metrics
stats = metrics.get_metrics()
print(f"Total: {stats['total_processed']}")
print(f"Errors: {stats['total_errors']}")
print(f"Avg time: {stats['avg_processing_time']:.3f}s")

ThrottleMiddleware

class flow.ThrottleMiddleware(delay_seconds)[source]

Bases: Middleware

Middleware that adds delays for rate limiting.

Examples

Create with 100ms delay:

>>> throttle = ThrottleMiddleware(0.1)
>>> throttle.delay_seconds
0.1

Use in a flow to limit processing rate:

>>> import asyncio
>>> import time
>>> from flow import flow
>>> async def example():
...     throttle = ThrottleMiddleware(0.05)  # 50ms delay
...     results = []
...
...     start = time.time()
...     await (
...         flow()
...         .source([1, 2], int)
...         .with_middleware(throttle)
...         .sink(results.append)
...         .execute(duration=1.0)
...     )
...     elapsed = time.time() - start
...
...     return len(results), elapsed > 0.1  # Should take > 100ms
>>> count, slow_enough = asyncio.run(example())
>>> count
2
>>> slow_enough
True

Parameters:: delay_seconds (float)

__init__(delay_seconds)[source]

Parameters:: delay_seconds (float)

async process(context, next_middleware)[source]

Process the context and call next middleware in chain.

Parameters:

context (ProcessingContext)
next_middleware (Callable[[ProcessingContext], Awaitable[Any]])

Return type:

Any

Example usage:

from flow import flow, ThrottleMiddleware

# Rate limit to 1 request per second
throttle = ThrottleMiddleware(delay_seconds=1.0)

builder.with_middleware(throttle)

RetryMiddleware

class flow.RetryMiddleware(max_attempts=3, backoff=1.0)[source]

Bases: Middleware

Middleware that retries failed operations.

Examples

Create with custom settings:

>>> retry = RetryMiddleware(max_attempts=5, backoff=0.1)
>>> retry.max_attempts
5
>>> retry.backoff
0.1

Usage in a flow:

>>> import asyncio
>>> from flow import flow
>>> # The retry middleware can be used to handle transient failures
>>> retry = RetryMiddleware(max_attempts=3, backoff=0.1)
>>> # In practice, it would retry operations that fail temporarily
>>> # For example, network requests or database operations

Parameters:

max_attempts (int, default: 3)
backoff (float, default: 1.0)

__init__(max_attempts=3, backoff=1.0)[source]

Parameters:

max_attempts (int, default: 3)
backoff (float, default: 1.0)

async process(context, next_middleware)[source]

Process the context and call next middleware in chain.

Parameters:

context (ProcessingContext)
next_middleware (Callable[[ProcessingContext], Awaitable[Any]])

Return type:

Any

Example usage:

from flow import flow, RetryMiddleware

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

builder.with_middleware(retry)

Custom Middleware

Creating custom middleware is straightforward:

Simple Example

class TimingMiddleware(Middleware):
    def __init__(self):
        self.timings = {}

    async def process(self, context, next_middleware):
        start = time.time()

        try:
            result = await next_middleware(context)
            elapsed = time.time() - start

            # Record timing
            if context.node_name not in self.timings:
                self.timings[context.node_name] = []
            self.timings[context.node_name].append(elapsed)

            return result
        except Exception:
            elapsed = time.time() - start
            print(f"Failed after {elapsed:.2f}s")
            raise

Advanced Example

class CachingMiddleware(Middleware):
    def __init__(self, ttl_seconds: float = 60.0):
        self.cache = {}
        self.ttl = ttl_seconds

    async def process(self, context, next_middleware):
        # Only cache transforms
        if context.node_type != 'transform':
            return await next_middleware(context)

        # Create cache key
        cache_key = (context.node_name, str(context.input_value))

        # Check cache
        if cache_key in self.cache:
            cached_value, timestamp = self.cache[cache_key]
            if time.time() - timestamp < self.ttl:
                print(f"Cache hit for {context.node_name}")
                return cached_value

        # Process and cache
        result = await next_middleware(context)
        self.cache[cache_key] = (result, time.time())

        return result

Middleware Composition

Middleware execution order matters:

# Execution order: logger -> retry -> throttle -> node
builder.with_middleware(logger, retry, throttle)

The first middleware in the list wraps all others:

logger = LoggingMiddleware()
retry = RetryMiddleware(max_attempts=3)
throttle = ThrottleMiddleware(delay_seconds=0.1)

# Logger sees everything, including retries
# Retry wraps throttle and node
# Throttle only affects successful operations
builder.with_middleware(logger, retry, throttle)

Best Practices

Single Responsibility: Each middleware should have one clear purpose
Minimal Overhead: Keep middleware lightweight
Error Propagation: Decide whether to handle or propagate errors
Context Metadata: Use metadata to pass information between middleware
Conditional Processing: Check node type/name for selective application

Example: Production Pipeline

from flow import flow, LoggingMiddleware, MetricsMiddleware, RetryMiddleware

# Production-ready pipeline with multiple concerns
logger = LoggingMiddleware(log_errors=True)
metrics = MetricsMiddleware()
retry = RetryMiddleware(max_attempts=3, backoff=1.0)

await (
    flow("Production Pipeline")
    .with_middleware(logger, metrics, retry)
    .source(data_source, dict)
    .transform(validate_data, dict)
    .transform(enrich_data, dict)
    .filter(is_valid)
    .to(save_to_database)
    .execute()
)

# Check metrics
stats = metrics.get_metrics()
if stats['total_errors'] > 0:
    alert_ops_team(stats)