# Common Patterns

This guide covers common patterns and best practices for building production-ready flows with Flokkit Flow.

## ETL Pipeline Pattern

Extract, Transform, Load (ETL) is one of the most common patterns:

```python
async def etl_pipeline():
    """Classic ETL pattern with error handling and monitoring."""
    builder = flow("ETL Pipeline", auto_stop=True)
    
    # Extract
    async def extract_from_source():
        async with aiohttp.ClientSession() as session:
            async with session.get("https://api.example.com/data") as response:
                data = await response.json()
                for record in data['records']:
                    yield record
    
    # Transform
    def transform_record(record: dict) -> dict:
        return {
            'id': record['id'],
            'name': record['name'].upper(),
            'value': record['value'] * 1.1,  # Add 10% markup
            'processed_at': datetime.now().isoformat()
        }
    
    # Validate
    def validate_record(record: dict) -> bool:
        required_fields = ['id', 'name', 'value']
        return all(field in record for field in required_fields)
    
    # Load
    async def load_to_database(record: dict):
        async with db_pool.acquire() as conn:
            await conn.execute(
                """INSERT INTO processed_records (id, name, value, processed_at)
                   VALUES ($1, $2, $3, $4)
                   ON CONFLICT (id) DO UPDATE SET
                   name = $2, value = $3, processed_at = $4""",
                record['id'], record['name'], record['value'], record['processed_at']
            )
    
    # Build pipeline with monitoring
    (builder
        .source(extract_from_source)
        .tap(lambda x: logger.info(f"Extracted: {x['id']}"))
        .to(transform_record)
        .filter(validate_record)
        .tap(lambda x: metrics.increment('records.processed'))
        .to(load_to_database, queue_size=50))
    
    await builder.run()
```

## Stream Processing Pattern

For continuous data streams:

```python
class StreamProcessor:
    """Real-time stream processing with windowing."""
    
    def __init__(self, window_size: int = 100):
        self.window = []
        self.window_size = window_size
    
    async def create_pipeline(self):
        builder = flow("Stream Processor", auto_stop=False)
        
        # Kafka consumer as source
        async def consume_events():
            consumer = AIOKafkaConsumer(
                'events',
                bootstrap_servers='localhost:9092',
                value_deserializer=lambda m: json.loads(m.decode('utf-8'))
            )
            await consumer.start()
            
            try:
                async for msg in consumer:
                    yield msg.value
            finally:
                await consumer.stop()
        
        # Window aggregation
        async def aggregate_window(event: dict) -> Optional[dict]:
            self.window.append(event)
            
            if len(self.window) >= self.window_size:
                # Calculate aggregates
                result = {
                    'window_start': self.window[0]['timestamp'],
                    'window_end': self.window[-1]['timestamp'],
                    'count': len(self.window),
                    'sum': sum(e['value'] for e in self.window),
                    'avg': sum(e['value'] for e in self.window) / len(self.window)
                }
                
                # Clear window
                self.window = []
                return result
            
            return None  # No output until window is full
        
        # Build pipeline
        (builder
            .source(consume_events)
            .to(aggregate_window)
            .filter(lambda x: x is not None)  # Only emit complete windows
            .to(self.publish_aggregates))
        
        return builder
    
    async def publish_aggregates(self, aggregate: dict):
        """Publish to Redis for real-time dashboard."""
        await redis.publish('aggregates', json.dumps(aggregate))
```

## Fan-Out/Fan-In Pattern

Process data in parallel paths:

```python
async def parallel_processing():
    """Fan-out to multiple processors, then fan-in results."""
    builder = flow("Parallel Processor")
    
    # Source
    source = builder.source(fetch_images)
    
    # Fan-out to parallel processors
    thumbnail_path = source.to(create_thumbnail, name="thumbnail")
    metadata_path = source.to(extract_metadata, name="metadata")
    analysis_path = source.to(analyze_content, name="analysis")
    
    # Process results separately
    thumbnail_path.to(save_thumbnail)
    metadata_path.to(save_metadata)
    
    # Fan-in: Merge analysis results
    analysis_path.to(enrich_with_ml_tags).to(save_analysis)
    
    # Alternative: Combine all results
    def combine_results(thumbnail: Image, metadata: dict, analysis: dict) -> dict:
        return {
            'thumbnail_url': upload_image(thumbnail),
            'metadata': metadata,
            'analysis': analysis
        }
    
    # Would need custom merge node for multiple typed inputs
    await builder.run()
```

## Request-Reply Pattern

Handle request-response workflows:

```python
class RequestReplyHandler:
    """Process requests and send replies."""
    
    def __init__(self):
        self.pending_requests = {}
    
    async def create_pipeline(self):
        builder = flow("Request Handler", auto_stop=False)
        
        # Receive requests
        async def receive_requests():
            async for request in request_queue:
                # Store request context
                request_id = str(uuid.uuid4())
                self.pending_requests[request_id] = request['reply_to']
                
                yield {
                    'id': request_id,
                    'data': request['data']
                }
        
        # Process request
        async def process_request(request: dict) -> dict:
            result = await complex_processing(request['data'])
            return {
                'id': request['id'],
                'result': result
            }
        
        # Send reply
        async def send_reply(response: dict):
            reply_to = self.pending_requests.pop(response['id'])
            await reply_queue.put({
                'correlation_id': response['id'],
                'destination': reply_to,
                'result': response['result']
            })
        
        (builder
            .source(receive_requests)
            .to(process_request, queue_size=100)
            .to(send_reply))
        
        return builder
```

## Batch Processing Pattern

Accumulate and process in batches:

```python
class BatchProcessor:
    """Efficient batch processing with timeout."""
    
    def __init__(self, batch_size: int = 1000, timeout: float = 5.0):
        self.batch_size = batch_size
        self.timeout = timeout
        self.current_batch = []
        self.last_batch_time = time.time()
    
    async def create_pipeline(self):
        builder = flow("Batch Processor")
        
        # Batch accumulator
        async def accumulate_batch(item: dict) -> Optional[List[dict]]:
            self.current_batch.append(item)
            now = time.time()
            
            # Emit batch if size reached or timeout
            if (len(self.current_batch) >= self.batch_size or 
                now - self.last_batch_time >= self.timeout):
                
                batch = self.current_batch[:]
                self.current_batch = []
                self.last_batch_time = now
                return batch
            
            return None
        
        # Bulk processor
        async def process_batch(batch: List[dict]):
            if not batch:
                return
            
            # Efficient bulk operation
            async with db_pool.acquire() as conn:
                await conn.copy_records_to_table(
                    'processed_items',
                    records=[(r['id'], r['data']) for r in batch],
                    columns=['id', 'data']
                )
            
            logger.info(f"Processed batch of {len(batch)} items")
        
        (builder
            .source(item_stream)
            .to(accumulate_batch)
            .filter(lambda x: x is not None)
            .to(process_batch, queue_size=10))
        
        return builder
```

## Circuit Breaker Pattern

Protect against cascading failures:

```python
class CircuitBreakerPipeline:
    """Pipeline with circuit breaker for external services."""
    
    def __init__(self):
        self.circuit_breaker = CircuitBreaker(
            failure_threshold=5,
            recovery_timeout=60,
            expected_exception=aiohttp.ClientError
        )
    
    async def create_pipeline(self):
        builder = flow("Protected Pipeline")
        
        # Protected external call
        @self.circuit_breaker
        async def call_external_api(data: dict) -> dict:
            async with aiohttp.ClientSession() as session:
                async with session.post(
                    "https://api.example.com/process",
                    json=data,
                    timeout=aiohttp.ClientTimeout(total=5)
                ) as response:
                    response.raise_for_status()
                    return await response.json()
        
        # Fallback when circuit is open
        def fallback_processor(data: dict) -> dict:
            logger.warning("Using fallback processor")
            return {
                'id': data['id'],
                'result': 'fallback',
                'processed_locally': True
            }
        
        # Wrapper that uses fallback
        async def protected_process(data: dict) -> dict:
            try:
                return await call_external_api(data)
            except CircuitBreakerError:
                return fallback_processor(data)
        
        (builder
            .source(data_stream)
            .to(protected_process)
            .to(save_result))
        
        return builder
```

## Saga Pattern

Manage distributed transactions:

```python
class SagaOrchestrator:
    """Implement saga pattern for distributed transactions."""
    
    async def create_order_saga(self):
        builder = flow("Order Saga")
        
        # Step 1: Reserve inventory
        async def reserve_inventory(order: dict) -> dict:
            try:
                reservation = await inventory_service.reserve(
                    order['item_id'], 
                    order['quantity']
                )
                return {**order, 'reservation_id': reservation['id']}
            except Exception as e:
                # Compensation not needed - nothing to undo
                raise
        
        # Step 2: Charge payment
        async def charge_payment(order: dict) -> dict:
            try:
                charge = await payment_service.charge(
                    order['customer_id'],
                    order['amount']
                )
                return {**order, 'charge_id': charge['id']}
            except Exception as e:
                # Compensate: Cancel reservation
                await inventory_service.cancel_reservation(
                    order['reservation_id']
                )
                raise
        
        # Step 3: Create shipment
        async def create_shipment(order: dict) -> dict:
            try:
                shipment = await shipping_service.create(
                    order['customer_id'],
                    order['item_id']
                )
                return {**order, 'shipment_id': shipment['id'], 'status': 'completed'}
            except Exception as e:
                # Compensate: Refund and cancel
                await payment_service.refund(order['charge_id'])
                await inventory_service.cancel_reservation(
                    order['reservation_id']
                )
                raise
        
        (builder
            .source(order_stream)
            .to(reserve_inventory)
            .to(charge_payment)
            .to(create_shipment)
            .to(save_completed_order))
        
        return builder
```

## Content-Based Router Pattern

Route messages based on content:

```python
async def content_router():
    """Route messages to different processors based on content."""
    builder = flow("Content Router")
    
    # Routing logic
    def route_message(message: dict) -> dict:
        msg_type = message.get('type', 'unknown')
        
        if msg_type == 'order':
            return process_order(message)
        elif msg_type == 'inventory':
            return process_inventory(message)
        elif msg_type == 'customer':
            return process_customer(message)
        else:
            logger.warning(f"Unknown message type: {msg_type}")
            return None
    
    # Type-specific processors
    async def process_order(msg: dict) -> dict:
        # Order-specific logic
        return {**msg, 'processed': 'order'}
    
    async def process_inventory(msg: dict) -> dict:
        # Inventory-specific logic
        return {**msg, 'processed': 'inventory'}
    
    async def process_customer(msg: dict) -> dict:
        # Customer-specific logic
        return {**msg, 'processed': 'customer'}
    
    (builder
        .source(message_queue)
        .to(route_message)
        .filter(lambda x: x is not None)
        .to(publish_result))
    
    await builder.run()
```

## Monitoring and Observability Pattern

Add comprehensive monitoring:

```python
class MonitoredPipeline:
    """Pipeline with full observability."""
    
    def __init__(self):
        self.metrics = MetricsCollector()
        self.tracer = Tracer()
    
    async def create_pipeline(self):
        builder = flow("Monitored Pipeline")
        
        # Wrap processing with tracing
        async def traced_process(item: dict) -> dict:
            with self.tracer.span("process_item") as span:
                span.set_attribute("item.id", item['id'])
                span.set_attribute("item.type", item['type'])
                
                start_time = time.time()
                try:
                    result = await process_item(item)
                    
                    # Record metrics
                    self.metrics.increment('items.processed')
                    self.metrics.histogram(
                        'processing.duration',
                        time.time() - start_time
                    )
                    
                    return result
                except Exception as e:
                    span.record_exception(e)
                    self.metrics.increment('items.failed')
                    raise
        
        # Health check tap
        def health_check(item: dict):
            self.metrics.gauge('pipeline.queue_size', 
                             get_current_queue_size())
            self.metrics.set('pipeline.last_item_time', 
                           time.time())
        
        (builder
            .source(data_source)
            .tap(lambda x: self.metrics.increment('items.received'))
            .to(traced_process)
            .tap(health_check)
            .to(save_result))
        
        return builder
```

## Testing Pattern

Structure pipelines for testability:

```python
class TestableProcessor:
    """Processor designed for easy testing."""
    
    def __init__(self, external_service=None):
        self.external_service = external_service or RealService()
    
    async def process(self, item: dict) -> dict:
        # Easy to mock external service
        enriched = await self.external_service.enrich(item)
        return self.transform(enriched)
    
    def transform(self, data: dict) -> dict:
        # Pure function - easy to test
        return {
            'id': data['id'],
            'value': data['value'] * 2,
            'timestamp': datetime.now().isoformat()
        }

# Test
async def test_processor():
    # Mock external service
    mock_service = Mock()
    mock_service.enrich.return_value = {'id': 1, 'value': 10}
    
    processor = TestableProcessor(mock_service)
    result = await processor.process({'id': 1})
    
    assert result['value'] == 20
    mock_service.enrich.assert_called_once()
```

## Best Practices Summary

1. **Use Type Hints**: Enable static type checking
2. **Handle Errors Gracefully**: Use lifecycle hooks
3. **Monitor Everything**: Add metrics and logging
4. **Design for Testing**: Keep logic pure and mockable
5. **Manage Resources**: Clean up in lifecycle hooks
6. **Control Backpressure**: Configure queues appropriately
7. **Document Patterns**: Make code self-documenting

## Next Steps

- Review [Examples](../examples/index.md) for complete implementations
- Check the [API Reference](../api/core.rst) for detailed documentation
- Join the community to share your patterns