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:
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:
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:
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:
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:
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:
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:
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:
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:
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:
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
Use Type Hints: Enable static type checking
Handle Errors Gracefully: Use lifecycle hooks
Monitor Everything: Add metrics and logging
Design for Testing: Keep logic pure and mockable
Manage Resources: Clean up in lifecycle hooks
Control Backpressure: Configure queues appropriately
Document Patterns: Make code self-documenting
Next Steps
Review Examples for complete implementations
Check the API Reference for detailed documentation
Join the community to share your patterns