Native Backend

pip install nadoo-flow-core

from nadoo_flow import WorkflowExecutor, create_workflow

# Create workflow
workflow = create_workflow({
    "id": "example",
    "nodes": [...],
    "edges": [...]
})

# Execute with native backend (default)
executor = WorkflowExecutor()
result = await executor.execute(workflow, {"input": "data"})

from nadoo_flow import ExecutionConfig, WorkflowExecutor

config = ExecutionConfig(
    # Parallel execution
    max_parallel_nodes=10,

    # Timeouts
    timeout_seconds=300,
    node_timeout=30,

    # Resilience
    retry_failed_nodes=True,
    max_retries=3,
    retry_delay=1.0,

    # Checkpointing
    enable_checkpoints=True,
    checkpoint_interval=5,

    # Tracing
    trace_execution=True,
    collect_metrics=True
)

executor = WorkflowExecutor(config)

from nadoo_flow.memory import ConversationMemory

# Configure memory
memory = ConversationMemory(
    max_messages=100,
    summarize_after=50,
    window_size=10
)

# Attach to workflow
workflow_context = WorkflowContext("workflow_1")
workflow_context.memory = memory

from nadoo_flow import SequentialExecutor

executor = SequentialExecutor()
result = await executor.execute(workflow, input_data)

from nadoo_flow import ParallelExecutor

executor = ParallelExecutor(max_workers=10)
result = await executor.execute(workflow, input_data)

from nadoo_flow import ConditionalExecutor

executor = ConditionalExecutor()
result = await executor.execute(workflow, input_data)

# Enable caching
config = ExecutionConfig(
    enable_cache=True,
    cache_ttl=3600,
    cache_size=1000
)

# Configure connection pools
config = ExecutionConfig(
    db_pool_size=20,
    http_pool_size=100,
    redis_pool_size=50
)

# Optimize async execution
config = ExecutionConfig(
    event_loop_policy="uvloop",  # Use uvloop for better performance
    max_concurrent_tasks=50,
    batch_size=100
)

class CheckpointedExecutor(WorkflowExecutor):
    """Executor with checkpointing support"""

    async def execute(self, workflow, input_data):
        # Enable checkpointing
        self.config.enable_checkpoints = True

        # Set checkpoint storage
        self.checkpoint_store = FileCheckpointStore("/tmp/checkpoints")

        # Execute with automatic checkpointing
        return await super().execute(workflow, input_data)

    async def resume_from_checkpoint(self, checkpoint_id):
        """Resume execution from checkpoint"""
        checkpoint = await self.checkpoint_store.load(checkpoint_id)
        return await self.continue_execution(checkpoint)

# Register custom node types with native backend
from nadoo_flow.backends.native import NodeRegistry

@NodeRegistry.register("custom_type")
class CustomNode(BaseNode):
    async def execute(self, node_context, workflow_context):
        # Custom execution logic
        return NodeResult(success=True)

class HookedExecutor(WorkflowExecutor):
    """Executor with lifecycle hooks"""

    async def pre_workflow_hook(self, workflow, input_data):
        """Called before workflow execution"""
        logger.info(f"Starting workflow {workflow.id}")

    async def post_node_hook(self, node, result):
        """Called after each node"""
        logger.info(f"Node {node.id} completed: {result.success}")

    async def post_workflow_hook(self, workflow, result):
        """Called after workflow completion"""
        logger.info(f"Workflow {workflow.id} finished")

# Enable detailed tracing
config = ExecutionConfig(
    trace_execution=True,
    trace_level="DEBUG",
    trace_output="workflow_trace.json"
)

executor = WorkflowExecutor(config)
result = await executor.execute(workflow, input_data)

# Access trace
trace = result.execution_trace
print(f"Total duration: {trace.duration}ms")
print(f"Node timings: {trace.node_timings}")

from nadoo_flow.metrics import MetricsCollector

# Configure metrics
metrics = MetricsCollector()
executor = WorkflowExecutor(metrics_collector=metrics)

# Execute
result = await executor.execute(workflow, input_data)

# Get metrics
stats = metrics.get_statistics()
print(f"Average node execution: {stats['avg_node_time']}ms")
print(f"Total workflows: {stats['total_workflows']}")

# Enable debug mode
config = ExecutionConfig(
    debug_mode=True,
    break_on_error=True,
    verbose_logging=True
)

# Debug execution
async with DebugExecutor(config) as executor:
    result = await executor.execute(workflow, input_data)

    # Inspect state at any point
    state = executor.get_current_state()
    print(f"Current node: {state.current_node}")
    print(f"Context: {state.context}")

# Configure error handling
config = ExecutionConfig(
    on_error="continue",  # continue, stop, retry
    error_handler=custom_error_handler,
    max_errors=5
)

class ErrorReportingExecutor(WorkflowExecutor):
    async def handle_error(self, error, node, context):
        """Custom error handling"""
        # Log error
        logger.error(f"Error in {node.id}: {error}")

        # Report to monitoring
        await self.report_to_sentry(error, node, context)

        # Attempt recovery
        if self.can_recover(error):
            return await self.recover(node, context)

        raise error