Middleware API
==============

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

Core Classes
------------

Middleware Base Class
~~~~~~~~~~~~~~~~~~~~~

.. autoclass:: flow.Middleware
   :members:
   :special-members: __init__
   :show-inheritance:

All middleware must implement the ``process`` method:

.. code-block:: python

   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
~~~~~~~~~~~~~~~~~

.. autoclass:: flow.ProcessingContext
   :members:
   :special-members: __init__
   :show-inheritance:

The context provides information about the current processing:

.. code-block:: python

   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
~~~~~~~~~~~~~~~~~

.. autoclass:: flow.LoggingMiddleware
   :members:
   :special-members: __init__
   :show-inheritance:

Example usage:

.. code-block:: python

   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
~~~~~~~~~~~~~~~~~

.. autoclass:: flow.MetricsMiddleware
   :members:
   :special-members: __init__
   :show-inheritance:

Example usage:

.. code-block:: python

   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
~~~~~~~~~~~~~~~~~~

.. autoclass:: flow.ThrottleMiddleware
   :members:
   :special-members: __init__
   :show-inheritance:

Example usage:

.. code-block:: python

   from flow import flow, ThrottleMiddleware
   
   # Rate limit to 1 request per second
   throttle = ThrottleMiddleware(delay_seconds=1.0)
   
   builder.with_middleware(throttle)

RetryMiddleware
~~~~~~~~~~~~~~~

.. autoclass:: flow.RetryMiddleware
   :members:
   :special-members: __init__
   :show-inheritance:

Example usage:

.. code-block:: python

   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
~~~~~~~~~~~~~~

.. code-block:: python

   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
~~~~~~~~~~~~~~~~

.. code-block:: python

   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:

.. code-block:: python

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

The first middleware in the list wraps all others:

.. code-block:: python

   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
--------------

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

Example: Production Pipeline
~~~~~~~~~~~~~~~~~~~~~~~~~~~~

.. code-block:: python

   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)