Best Practices
This guide outlines best practices for developing with the Kailash Python SDK.
Node Development
Creating Custom Nodes
When creating custom nodes, follow these patterns:
from kailash.nodes.base import Node
from typing import Dict, Any, Optional
from pydantic import BaseModel, Field
class MyCustomNodeConfig(BaseModel):
"""Configuration schema for MyCustomNode."""
parameter: str = Field(..., description="Required parameter")
optional_param: Optional[int] = Field(None, description="Optional parameter")
class MyCustomNode(Node):
"""
Custom node for specific business logic.
Example:
>>> from kailash.nodes.base import Node
>>> node = MyCustomNode(
... node_id="custom_1",
... config={"parameter": "value"}
... )
>>> result = node.execute({"input": "data"})
"""
def __init__(self, node_id: str, config: Dict[str, Any]):
super().__init__(node_id, config)
self.config_model = MyCustomNodeConfig(**config)
def execute(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
"""Execute the node logic."""
try:
# Validate inputs
self.validate_inputs(inputs)
# Process data
result = self._process_data(inputs)
# Return structured output
return {
"success": True,
"data": result,
"metadata": {"processed_at": "2024-01-01T00:00:00Z"}
}
except Exception as e:
return {
"success": False,
"error": str(e),
"metadata": {"error_type": type(e).__name__}
}
Key Principles:
Use Pydantic for Configuration: Define configuration schemas using Pydantic models
Input Validation: Always validate inputs before processing
Structured Output: Return consistent output formats with success indicators
Error Handling: Catch and handle exceptions gracefully
Documentation: Include comprehensive docstrings with examples
Data Handling
File Processing:
import pandas as pd
from pathlib import Path
def safe_file_read(file_path: str) -> pd.DataFrame:
"""Safely read CSV files with error handling."""
try:
path = Path(file_path)
if not path.exists():
raise FileNotFoundError(f"File not found: {file_path}")
return pd.read_csv(path)
except Exception as e:
raise ValueError(f"Failed to read file {file_path}: {e}")
Memory Management:
import pandas as pd
def process_large_dataset(data: pd.DataFrame, chunk_size: int = 10000):
"""Process large datasets in chunks to manage memory."""
for i in range(0, len(data), chunk_size):
chunk = data[i:i + chunk_size]
yield process_chunk(chunk)
Workflow Design
Workflow Patterns
Linear Processing:
from kailash.workflow.builder import WorkflowBuilder
def create_linear_workflow():
"""Create a simple linear processing workflow."""
builder = WorkflowBuilder()
# Add nodes in sequence
builder.add_node("reader", "CSVReaderNode", {"file_path": "input.csv"})
builder.add_node("processor", "DataProcessor", {"operation": "clean"})
builder.add_node("writer", "CSVWriterNode", {"file_path": "output.csv"})
# Connect nodes
builder.add_edge("reader", "processor")
builder.add_edge("processor", "writer")
return builder.build()
Parallel Processing:
def create_parallel_workflow():
"""Create workflow with parallel processing branches."""
builder = WorkflowBuilder()
# Input node
builder.add_node("input", "CSVReaderNode", {"file_path": "data.csv"})
# Parallel processing branches
builder.add_node("process_a", "ProcessorA", {})
builder.add_node("process_b", "ProcessorB", {})
# Merge results
builder.add_node("merge", "DataMerger", {})
# Connect parallel branches
builder.add_edge("input", "process_a")
builder.add_edge("input", "process_b")
builder.add_edge("process_a", "merge")
builder.add_edge("process_b", "merge")
return builder.build()
Conditional Routing:
def create_conditional_workflow():
"""Create workflow with conditional logic."""
builder = WorkflowBuilder()
builder.add_node("input", "CSVReaderNode", {})
builder.add_node("validator", "DataValidator", {})
builder.add_node("clean_data", "DataCleaner", {})
builder.add_node("reject_data", "DataRejector", {})
# Add conditional routing
builder.add_conditional_edge(
"validator",
condition=lambda x: x.get("valid", False),
true_target="clean_data",
false_target="reject_data"
)
return builder.build()
Error Handling
Node-Level Error Handling
from kailash.nodes.base import Node
class RobustNode(Node):
"""Node with comprehensive error handling."""
def execute(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
try:
# Validate inputs
self._validate_inputs(inputs)
# Process with retries
result = self._execute_with_retry(inputs)
return {"success": True, "data": result}
except ValidationError as e:
return {
"success": False,
"error": f"Input validation failed: {e}",
"error_type": "validation"
}
except Exception as e:
return {
"success": False,
"error": str(e),
"error_type": "execution"
}
def _execute_with_retry(self, inputs: Dict[str, Any], max_retries: int = 3):
"""Execute with retry logic."""
for attempt in range(max_retries):
try:
return self._core_logic(inputs)
except Exception as e:
if attempt == max_retries - 1:
raise
time.sleep(2 ** attempt) # Exponential backoff
Workflow-Level Error Handling
def create_error_resilient_workflow():
"""Create workflow with error handling and recovery."""
builder = WorkflowBuilder()
# Main processing path
builder.add_node("processor", "DataProcessor", {})
builder.add_node("validator", "ResultValidator", {})
# Error handling path
builder.add_node("error_handler", "ErrorHandler", {})
builder.add_node("fallback", "FallbackProcessor", {})
# Add error routing
builder.add_error_handler("processor", "error_handler")
builder.add_edge("error_handler", "fallback")
return builder.build()
Testing Strategies
Unit Testing Nodes
import pytest
from unittest.mock import Mock, patch
class TestMyCustomNode:
"""Test suite for MyCustomNode."""
def test_successful_execution(self):
"""Test successful node execution."""
node = MyCustomNode("test_node", {"parameter": "test"})
inputs = {"input": "test_data"}
result = node.execute(inputs)
assert result["success"] is True
assert "data" in result
def test_input_validation(self):
"""Test input validation."""
node = MyCustomNode("test_node", {"parameter": "test"})
invalid_inputs = {}
result = node.execute(invalid_inputs)
assert result["success"] is False
assert "error" in result
@patch('mymodule.external_api_call')
def test_with_mocked_dependencies(self, mock_api):
"""Test node with mocked external dependencies."""
mock_api.return_value = {"status": "ok"}
node = MyCustomNode("test_node", {"parameter": "test"})
result = node.execute({"input": "data"})
mock_api.assert_called_once()
assert result["success"] is True
Integration Testing Workflows
class TestWorkflowIntegration:
"""Integration tests for complete workflows."""
def test_end_to_end_workflow(self, tmp_path):
"""Test complete workflow execution."""
# Setup test data
input_file = tmp_path / "input.csv"
input_file.write_text("name,age\\nJohn,30\\nJane,25")
# Create workflow
workflow = create_linear_workflow()
# Execute workflow
result = workflow.execute({"input_file": str(input_file)})
# Verify results
assert result["success"] is True
assert (tmp_path / "output.csv").exists()
Performance Optimization
Memory Management
import gc
from memory_profiler import profile
@profile
def memory_efficient_processing(large_dataset):
"""Process large datasets efficiently."""
# Process in chunks
chunk_size = 10000
for i in range(0, len(large_dataset), chunk_size):
chunk = large_dataset[i:i + chunk_size]
# Process chunk
processed = process_chunk(chunk)
# Yield results immediately
yield processed
# Clean up references
del chunk, processed
gc.collect()
Parallel Execution
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import multiprocessing as mp
def parallel_node_execution(nodes, inputs):
"""Execute multiple nodes in parallel."""
# Use ThreadPoolExecutor for I/O-bound tasks
with ThreadPoolExecutor(max_workers=mp.cpu_count()) as executor:
futures = {
executor.submit(node.execute, inputs): node
for node in nodes
}
results = {}
for future in futures:
node = futures[future]
try:
results[node.node_id] = future.result(timeout=30)
except Exception as e:
results[node.node_id] = {"success": False, "error": str(e)}
return results
Caching Strategies
from functools import lru_cache
import hashlib
import pickle
from kailash.nodes.base import Node
class CacheableNode(Node):
"""Node with built-in caching capabilities."""
def __init__(self, node_id: str, config: Dict[str, Any]):
super().__init__(node_id, config)
self.cache = {}
def execute(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
"""Execute with caching."""
cache_key = self._generate_cache_key(inputs)
if cache_key in self.cache:
return self.cache[cache_key]
result = self._execute_logic(inputs)
self.cache[cache_key] = result
return result
def _generate_cache_key(self, inputs: Dict[str, Any]) -> str:
"""Generate cache key from inputs."""
serialized = pickle.dumps(inputs, protocol=pickle.HIGHEST_PROTOCOL)
return hashlib.md5(serialized).hexdigest()
Monitoring and Logging
Performance Monitoring
import time
import logging
from functools import wraps
def monitor_performance(func):
"""Decorator to monitor function performance."""
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
memory_before = get_memory_usage()
try:
result = func(*args, **kwargs)
execution_time = time.time() - start_time
memory_after = get_memory_usage()
logging.info(f"{func.__name__} completed in {execution_time:.2f}s")
logging.info(f"Memory usage: {memory_after - memory_before:.2f}MB")
return result
except Exception as e:
logging.error(f"{func.__name__} failed: {e}")
raise
return wrapper
Structured Logging
import logging
import json
from datetime import datetime
class StructuredLogger:
"""Structured logging for workflow execution."""
def __init__(self, name: str):
self.logger = logging.getLogger(name)
self.logger.setLevel(logging.INFO)
def log_node_execution(self, node_id: str, inputs: dict, outputs: dict):
"""Log node execution details."""
log_entry = {
"timestamp": datetime.utcnow().isoformat(),
"event": "node_execution",
"node_id": node_id,
"input_size": len(str(inputs)),
"output_size": len(str(outputs)),
"success": outputs.get("success", False)
}
self.logger.info(json.dumps(log_entry))
Security Considerations
Input Sanitization
import re
from pathlib import Path
def sanitize_file_path(file_path: str) -> str:
"""Sanitize file paths to prevent directory traversal."""
# Remove dangerous patterns
sanitized = re.sub(r'\.\./', '', file_path)
sanitized = re.sub(r'\.\.\\', '', sanitized)
# Ensure path is within allowed directory
path = Path(sanitized).resolve()
allowed_dir = Path("/allowed/directory").resolve()
if not str(path).startswith(str(allowed_dir)):
raise ValueError("Path outside allowed directory")
return str(path)
Configuration Validation
from pydantic import BaseModel, validator
class SecureNodeConfig(BaseModel):
"""Secure configuration with validation."""
api_key: str
file_path: str
@validator('api_key')
def validate_api_key(cls, v):
if len(v) < 32:
raise ValueError('API key too short')
return v
@validator('file_path')
def validate_file_path(cls, v):
return sanitize_file_path(v)
Cyclic Workflow Best Practices
Designing Cyclic Workflows
When building iterative workflows, follow these patterns:
from kailash import Workflow
from kailash.nodes.base_cycle_aware import CycleAwareNode
from typing import Dict, Any
class IterativeProcessorNode(CycleAwareNode):
"""Best practices for cycle-aware nodes."""
def run(self, context: Dict[str, Any], **kwargs) -> Dict[str, Any]:
# 1. Always get iteration info first
iteration = self.get_iteration(context)
prev_state = self.get_previous_state(context)
# 2. Handle first iteration gracefully
if iteration == 0:
# Initialize state
current_value = kwargs.get("initial_value", 0)
self.log_cycle_info(context, "Starting iterative process")
else:
# Use previous state
current_value = prev_state.get("value", 0)
# 3. Perform processing
new_value = self.process_value(current_value)
# 4. Track metrics for convergence
self.accumulate_values(context, "value", new_value)
trend = self.detect_convergence_trend(context, "value")
# 5. Save state for next iteration
self.set_cycle_state({
"value": new_value,
"history": prev_state.get("history", []) + [new_value]
})
# 6. Determine convergence with multiple criteria
converged = (
new_value > 0.95 or # Threshold reached
trend["converging"] or # Trending toward convergence
trend["plateau_detected"] or # No improvement
iteration >= self.max_iterations - 1 # Safety limit
)
# 7. Return structured output
return {
"value": new_value,
"converged": converged,
"iteration": iteration,
"improvement": new_value - current_value
}
Cycle Safety Guidelines
Always Set Convergence Conditions:
# Good: Multiple convergence criteria
workflow.create_cycle("processing_cycle") \
.connect("processor", "processor") \
.max_iterations(100) \
.converge_when("converged == True or error < 0.001") \
.build()
# Bad: No convergence check - NEVER DO THIS!
# workflow.connect("processor", "processor", cycle=True) # Dangerous!
Use Reasonable Iteration Limits:
# Consider the problem domain
workflow.create_cycle("optimizer_cycle") \
.connect("optimizer", "optimizer") \
.max_iterations(1000) \
.converge_when("loss < 0.01") \
.build()
workflow.create_cycle("retry_cycle") \
.connect("retry", "retry") \
.max_iterations(5) \
.converge_when("success == True") \
.build()
Implement Memory Management:
class MemoryEfficientCycleNode(CycleAwareNode):
def run(self, context, **kwargs):
# Limit history size
MAX_HISTORY = 100
prev_state = self.get_previous_state(context)
history = prev_state.get("history", [])
# Trim history to prevent memory growth
if len(history) > MAX_HISTORY:
history = history[-MAX_HISTORY:]
# Process and update
result = self.process_data(kwargs["data"])
history.append(result["metric"])
self.set_cycle_state({"history": history})
return result
Performance Optimization for Cycles
from kailash.runtime.parallel_cyclic import ParallelCyclicRuntime
# Use parallel runtime for independent cycles
runtime = ParallelCyclicRuntime(
max_workers=4,
cycle_batch_size=10 # Process multiple iterations in parallel
)
# Example: Parallel optimization with multiple starting points
workflow = Workflow("parallel_optimization")
# Add multiple optimizers
for i in range(4):
workflow.add_node(f"optimizer_{i}", OptimizerNode())
workflow.create_cycle(f"optimizer_{i}_cycle") \
.connect(f"optimizer_{i}", f"optimizer_{i}") \
.max_iterations(100) \
.converge_when("converged == True") \
.build()
# Execute all optimizers in parallel
results = runtime.execute(workflow, parameters={
f"optimizer_{i}": {"initial_value": i * 0.25}
for i in range(4)
})
Common Pitfalls to Avoid
Incorrect Edge Marking:
# Wrong: Marking all edges as cycles - DEPRECATED PATTERN
# workflow.connect("A", "B", cycle=True) # ❌
# workflow.connect("B", "C", cycle=True) # ❌
# workflow.connect("C", "A", cycle=True) # ❌
# Correct: Use CycleBuilder API
workflow.connect("A", "B") # ✓ Regular forward edge
workflow.connect("B", "C") # ✓ Regular forward edge
workflow.create_cycle("abc_cycle") \
.connect("C", "A") \
.max_iterations(50) \
.converge_when("converged == True") \
.build() # ✓ Cycle with proper configuration
Poor State Management:
# Wrong: Storing entire datasets in state
self.set_cycle_state({"full_dataset": large_dataframe}) # ❌
# Correct: Store only essential state
self.set_cycle_state({
"summary_stats": dataframe.describe().to_dict(),
"row_count": len(dataframe),
"quality_score": calculate_quality(dataframe)
}) # ✓
Missing Error Handling:
class RobustCycleNode(CycleAwareNode):
def run(self, context, **kwargs):
iteration = self.get_iteration(context)
try:
result = self.process_data(kwargs["data"])
return {"success": True, "result": result}
except Exception as e:
# Log error with context
self.log_cycle_info(context,
f"Error at iteration {iteration}: {str(e)}")
# Decide whether to continue
if iteration < 3: # Retry a few times
return {"success": False, "retry": True}
else:
# Give up and converge
return {"success": False, "error": str(e),
"converged": True}