Nodes

This section provides comprehensive documentation for all node types available in the Kailash SDK.

Base Node Classes 

Node 

class kailash.nodes.base.Node(**kwargs)[source]

Bases: ABC

Base class for all nodes in the Kailash system.

This abstract class defines the contract that all nodes must implement. It provides the foundation for:

Parameter validation and type checking
Execution lifecycle management
Error handling and reporting
Serialization for workflow export
Configuration management

Design Philosophy: - Nodes are stateless processors of data - All configuration is provided at initialization - Runtime inputs are validated against schemas - Outputs must be JSON-serializable - Errors are wrapped in appropriate exception types

Inheritance Pattern: All concrete nodes must:

Implement get_parameters() to define inputs
Implement run() to process data
Call super().__init__() with configuration
Use self.logger for logging

Upstream components: - Workflow: Creates and manages node instances - NodeRegistry: Provides node classes for instantiation - CLI/UI: Configures nodes based on user input

Downstream usage: - LocalRuntime: Executes nodes in workflows - TaskManager: Tracks node execution status - WorkflowExporter: Serializes nodes for export

classmethod __init_subclass__(**subclass_kwargs)[source]

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

__init__(**kwargs)[source]

Initialize the node with configuration parameters.

This method performs the following initialization steps:

Sets the node ID (defaults to class name)
Creates metadata from provided arguments
Sets up logging for the node
Stores configuration in self.config
Validates configuration against parameters

The configuration is validated by calling _validate_config(), which checks that all required parameters are present and of the correct type.

Parameters:: **kwargs – Configuration parameters including: - id: Optional custom node ID - name: Display name for the node - description: Node description - version: Node version - author: Node author - tags: Set of tags for discovery - Any parameters defined in get_parameters()
Raises:: NodeConfigurationError – If configuration is invalid or if metadata validation fails

Downstream effects:

Creates self.metadata for discovery
Sets up self.logger for execution logging
Stores self.config for runtime access
Validates parameters are correctly specified

get_workflow_context(key: str, default: Any | None = None) → Any[source]

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

set_workflow_context(key: str, value: Any) → None[source]

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

abstractmethod get_parameters() → dict[str, NodeParameter][source]

Define the parameters this node accepts.

This abstract method must be implemented by all concrete nodes to specify their input schema. The parameters define:

What inputs the node expects
Type requirements for each input
Whether inputs are required or optional
Default values for optional inputs
Documentation for each parameter

The returned dictionary is used throughout the node lifecycle:

During initialization: _validate_config() checks configuration
During execution: validate_inputs() validates runtime data
During workflow creation: Used for connection validation
During export: Included in workflow manifests

Example

>>> def get_parameters(self):
...     return {
...         'input_file': NodeParameter(
...             name='input_file',
...             type=str,
...             required=True,
...             description='Path to input CSV file'
...         ),
...         'delimiter': NodeParameter(
...             name='delimiter',
...             type=str,
...             required=False,
...             default=',',
...             description='CSV delimiter character'
...         )
...     }

Returns:: Dictionary mapping parameter names to their definitions
Return type:: dict[str, NodeParameter]

Used by:

_validate_config(): Validates configuration matches parameters
validate_inputs(): Validates runtime inputs
to_dict(): Includes parameters in serialization
Workflow.connect(): Validates compatible connections

get_output_schema() → dict[str, NodeParameter][source]

Define output parameters for this node.

This optional method allows nodes to specify their output schema for validation. If not overridden, outputs will only be validated for JSON-serializability.

Design purpose: - Enables static analysis of node outputs - Provides runtime validation of output types - Supports automatic documentation of outputs - Facilitates workflow validation and type checking

The output schema serves similar purposes as input parameters:

Type validation during execution
Documentation for downstream consumers
Workflow connection validation
Export manifest generation

Example

>>> def get_output_schema(self):
...     return {
...         'dataframe': NodeParameter(
...             name='dataframe',
...             type=dict,
...             required=True,
...             description='Processed data as dictionary'
...         ),
...         'row_count': NodeParameter(
...             name='row_count',
...             type=int,
...             required=True,
...             description='Number of rows processed'
...         ),
...         'processing_time': NodeParameter(
...             name='processing_time',
...             type=float,
...             required=False,
...             description='Time taken to process in seconds'
...         )
...     }

Returns:: Dictionary mapping output names to their parameter definitions Empty dict by default (no schema validation)
Return type:: dict[str, NodeParameter]

Used by:

validate_outputs(): Validates runtime outputs
Workflow.connect(): Validates connections between nodes
Documentation generators: Create output documentation
Export systems: Include output schemas in manifests

abstractmethod run(**kwargs) → dict[str, Any][source]

Execute the node’s logic.

This is the core method that implements the node’s data processing logic. It receives validated inputs and must return a dictionary of outputs.

Design requirements:

Must be stateless - no side effects between runs
All inputs are provided as keyword arguments
Must return a dictionary (JSON-serializable)
Should handle errors gracefully
Can use self.config for configuration values
Should use self.logger for status reporting

The method is called by execute() which handles:

Input validation before calling run()
Output validation after run() completes
Error wrapping and logging
Execution timing and metrics

Example

>>> def run(self, input_file, delimiter=','):
...     df = pd.read_csv(input_file, delimiter=delimiter)
...     return {
...         'dataframe': df.to_dict(),
...         'row_count': len(df),
...         'columns': list(df.columns)
...     }

Parameters:: **kwargs – Validated input parameters matching get_parameters()
Returns:: Dictionary of outputs that will be validated and passed to downstream nodes
Raises:: NodeExecutionError – If execution fails (will be caught and re-raised by execute())
Return type:: dict[str, Any]

Called by:

execute(): Wraps with validation and error handling
LocalRuntime: During workflow execution
TestRunner: During unit testing

validate_inputs(**kwargs) → dict[str, Any][source]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

Gets parameter definitions from get_parameters()
Checks each parameter for:
- Presence (if required)
- Type compatibility
- Null handling for optional parameters
Attempts type conversion if needed
Applies default values for missing optional parameters

Key behaviors:

Required parameters must be provided or have defaults
Optional parameters can be None
Type mismatches attempt conversion before failing
Error messages include parameter descriptions

Example flow:: # Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

Type conversions applied
Defaults for missing optional parameters
None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

execute(): Before passing inputs to run()
Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any][source]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

execute(**runtime_inputs) → dict[str, Any][source]

Execute the node with validation and error handling.

This is the main entry point for node execution that orchestrates the complete execution lifecycle:

Input validation (validate_inputs)
Execution (run)
Output validation (validate_outputs)
Error handling and logging
Performance metrics

Execution flow:

Logs execution start
Validates inputs against parameter schema
Calls run() with validated inputs
Validates outputs are JSON-serializable
Logs execution time
Returns validated outputs

Error handling strategy:

NodeValidationError: Re-raised as-is (input/output issues)
NodeExecutionError: Re-raised as-is (run() failures)
Other exceptions: Wrapped in NodeExecutionError

Performance tracking:

Records execution start/end times
Logs total execution duration
Includes timing in execution logs

Returns:

Dictionary of validated outputs from run()

Raises:

NodeExecutionError – If execution fails in run()
NodeValidationError – If input/output validation fails

Return type:

dict[str, Any]

Called by:

LocalRuntime: During workflow execution
TaskManager: With execution tracking
Unit tests: For node testing

Downstream effects:

Logs provide execution history
Metrics enable performance monitoring
Validation ensures data integrity

get_cache_stats() → dict[str, Any][source]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

clear_cache() → None[source]

Clear the parameter resolution cache and reset statistics.

Return type:: None

warm_cache(patterns: list[dict[str, Any]]) → None[source]

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

to_dict() → dict[str, Any][source]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

Workflow export
Node persistence
API responses
Configuration sharing

The serialized format includes:

id: Unique node identifier
type: Node class name
metadata: Complete node metadata
config: Current configuration
parameters: Parameter definitions with types

Type serialization:

Python types are converted to string names
Complex types may require custom handling
Parameter defaults are included

Returns:

Node identification and type
Complete metadata
Configuration values
Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

WorkflowExporter: For workflow serialization
CLI: For node inspection
API: For node information endpoints
Debugging: For node state inspection

AsyncNode 

class kailash.nodes.base_async.AsyncNode(**kwargs)[source]

Bases: EventEmitterMixin, SecurityMixin, PerformanceMixin, LoggingMixin, Node

Base class for asynchronous nodes with enterprise capabilities.

This class extends the standard Node class with: 1. Async execution capabilities 2. Event emission for monitoring 3. Security features (input validation, sanitization) 4. Performance monitoring (execution tracking) 5. Enhanced logging (structured logging with context)

Inherits from:: EventEmitterMixin: Async-compatible event emission for monitoring SecurityMixin: Security validation and input sanitization PerformanceMixin: Performance tracking and metrics collection LoggingMixin: Enhanced logging with context support Node: Base node functionality and validation

Use Cases: 1. API calls and network operations 2. Database queries 3. File operations 4. External service integrations 5. LLM/AI model inference

Design Philosophy: - Maintain backward compatibility with synchronous nodes - Support both sync and async execution methods - Provide enterprise-grade features through mixins - Clear error handling and logging for async operations - Enable efficient parallel execution in workflows

Mixin Order Rationale: - EventEmitterMixin first: Already async, no conflicts - SecurityMixin: May need methods from PerformanceMixin - PerformanceMixin: May need logging from LoggingMixin - LoggingMixin: May need Node methods - Node last: Base class with fundamental methods

Usage Pattern: - Override async_run() instead of run() for async functionality - All enterprise features automatically available - Use event emission for monitoring - Security validation automatic in execute_async()

Upstream components: - Workflow: Creates and manages node instances - AsyncWorkflowExecutor: Executes nodes in parallel where possible - AsyncLocalRuntime: Runs workflows with async support

Downstream usage: - Custom AsyncNodes: Implement async_run() for I/O-bound operations - TaskManager: Tracks node execution status

__init__(**kwargs)[source]

Initialize AsyncNode with all enterprise capabilities.

This calls the MRO chain to initialize all mixins and the base Node. The MRO ensures each mixin’s __init__ is called exactly once.

Parameters:: **kwargs – Configuration parameters for node and mixins - All Node parameters (node_id, node_type, config, etc.) - security_config: Optional SecurityConfig for SecurityMixin - log_level: Log level for LoggingMixin (default: “INFO”) - enable_performance_tracking: Enable performance metrics (default: True)

execute(**runtime_inputs) → dict[str, Any][source]

Execute the node synchronously by running async code with proper event loop handling.

This enhanced implementation handles all event loop scenarios: 1. No event loop: Create new one with asyncio.run() 2. Event loop running: Use ThreadPoolExecutor with isolated loop 3. Threaded contexts: Proper thread-safe execution 4. Windows compatibility: ProactorEventLoopPolicy support

Parameters:

**runtime_inputs – Runtime inputs for node execution

Returns:

Dictionary of validated outputs

Raises:

NodeValidationError – If inputs or outputs are invalid
NodeExecutionError – If execution fails

Return type:

dict[str, Any]

run(**kwargs) → dict[str, Any][source]

Synchronous run is not supported for AsyncNode.

AsyncNode subclasses should implement async_run() instead of run(). This method exists to provide a clear error message if someone accidentally tries to implement run() on an async node.

Raises:: NotImplementedError – Always, as async nodes must use async_run()
Return type:: dict[str, Any]

async async_run(**kwargs) → dict[str, Any][source]

Asynchronous execution method for the node.

This method should be overridden by subclasses to implement asynchronous execution logic. The default implementation raises NotImplementedError to ensure async nodes properly implement their async behavior.

Parameters:: **kwargs – Input parameters for node execution
Returns:: Dictionary of outputs matching the node’s output schema
Raises:: NodeExecutionError – If execution fails
Return type:: dict[str, Any]

async execute_async(**runtime_inputs) → dict[str, Any][source]

Execute the node asynchronously with validation and error handling.

This method follows the same pattern as execute() but supports asynchronous execution. It performs:

Input validation
Execution via async_run()
Output validation
Error handling and logging

Parameters:

**runtime_inputs – Runtime inputs for node execution

Returns:

Dictionary of validated outputs

Raises:

NodeValidationError – If inputs or outputs are invalid
NodeExecutionError – If execution fails

Return type:

dict[str, Any]

async audit_log(action: str, details: Dict[str, Any]) → None[source]

Log an audit event (async override).

Overrides SecurityMixin.audit_log to prevent blocking the event loop. Uses asyncio.to_thread() to offload print() to thread pool.

Parameters:

action (str) – Action being audited
details (Dict[str, Any]) – Additional details about the action

Return type:

None

async log_security_event(event: str, level: str = 'INFO') → None[source]

Log a security-related event (async override).

This method provides async logging for security events when audit logging is enabled in security_config.

Parameters:

event (str) – Description of the security event
level (str) – Log level (INFO, WARNING, ERROR)

Return type:

None

async validate_and_sanitize_inputs(inputs: Dict[str, Any]) → Dict[str, Any][source]

Validate and sanitize input parameters (async override).

Overrides SecurityMixin.validate_and_sanitize_inputs when the full SecurityMixin from mixins.py is used (with logging).

Parameters:: inputs (Dict[str, Any]) – Dictionary of input parameters
Returns:: Dictionary of validated and sanitized parameters
Return type:: Dict[str, Any]

async log_with_context(level: str, message: str, **context) → None[source]

Log a message with additional context (async override).

Overrides LoggingMixin.log_with_context to prevent blocking.

Parameters:

level (str) – Log level (debug, info, warning, error, critical)
message (str) – Log message
**context – Additional context to include

Return type:

None

async log_node_execution(operation: str, **context) → None[source]

Log node execution information (async override).

Overrides LoggingMixin.log_node_execution to prevent blocking.

Parameters:

operation (str) – Type of operation being performed
**context – Additional context

Return type:

None

async log_error_with_traceback(error: Exception, operation: str = 'unknown') → None[source]

Log an error with full traceback information (async override).

Overrides LoggingMixin.log_error_with_traceback to prevent blocking.

Parameters:

error (Exception) – Exception that occurred
operation (str) – Operation that failed

Return type:

None

async log_info(message: str, **extra) → None[source]

Log info message with context (async override).

Overrides LoggingMixin.log_info to prevent blocking.

Parameters:

message (str) – Log message
**extra – Additional context

Return type:

None

async log_error(message: str, error: Exception | None = None, **extra) → None[source]

Log error message with context (async override).

Overrides LoggingMixin.log_error to prevent blocking.

Parameters:

message (str) – Log message
error (Exception | None) – Optional exception to include
**extra – Additional context

Return type:

None

async log_warning(message: str, **extra) → None[source]

Log warning message with context (async override).

Overrides LoggingMixin.log_warning to prevent blocking.

Parameters:

message (str) – Log message
**extra – Additional context

Return type:

None

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:: None

async emit_node_completed(outputs: Dict[str, Any] | None = None, execution_time_ms: float | None = None)

Emit node completed event.

Parameters:

outputs (Dict[str, Any] | None)
execution_time_ms (float | None)

async emit_node_failed(error: str)

Emit node failed event.

Parameters:: error (str)

async emit_node_progress(progress_percent: float, message: str | None = None)

Emit node progress event.

Parameters:

progress_percent (float)
message (str | None)

async emit_node_started(inputs: Dict[str, Any] | None = None)

Emit node started event.

Parameters:: inputs (Dict[str, Any] | None)

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_output_schema() → dict[str, NodeParameter]

Define output parameters for this node.

This optional method allows nodes to specify their output schema for validation. If not overridden, outputs will only be validated for JSON-serializability.

Design purpose: - Enables static analysis of node outputs - Provides runtime validation of output types - Supports automatic documentation of outputs - Facilitates workflow validation and type checking

The output schema serves similar purposes as input parameters:

Type validation during execution
Documentation for downstream consumers
Workflow connection validation
Export manifest generation

Example

>>> def get_output_schema(self):
...     return {
...         'dataframe': NodeParameter(
...             name='dataframe',
...             type=dict,
...             required=True,
...             description='Processed data as dictionary'
...         ),
...         'row_count': NodeParameter(
...             name='row_count',
...             type=int,
...             required=True,
...             description='Number of rows processed'
...         ),
...         'processing_time': NodeParameter(
...             name='processing_time',
...             type=float,
...             required=False,
...             description='Time taken to process in seconds'
...         )
...     }

Returns:: Dictionary mapping output names to their parameter definitions Empty dict by default (no schema validation)
Return type:: dict[str, NodeParameter]

Used by:

validate_outputs(): Validates runtime outputs
Workflow.connect(): Validates connections between nodes
Documentation generators: Create output documentation
Export systems: Include output schemas in manifests

abstractmethod get_parameters() → dict[str, NodeParameter]

Define the parameters this node accepts.

This abstract method must be implemented by all concrete nodes to specify their input schema. The parameters define:

What inputs the node expects
Type requirements for each input
Whether inputs are required or optional
Default values for optional inputs
Documentation for each parameter

The returned dictionary is used throughout the node lifecycle:

During initialization: _validate_config() checks configuration
During execution: validate_inputs() validates runtime data
During workflow creation: Used for connection validation
During export: Included in workflow manifests

Example

>>> def get_parameters(self):
...     return {
...         'input_file': NodeParameter(
...             name='input_file',
...             type=str,
...             required=True,
...             description='Path to input CSV file'
...         ),
...         'delimiter': NodeParameter(
...             name='delimiter',
...             type=str,
...             required=False,
...             default=',',
...             description='CSV delimiter character'
...         )
...     }

Returns:: Dictionary mapping parameter names to their definitions
Return type:: dict[str, NodeParameter]

Used by:

_validate_config(): Validates configuration matches parameters
validate_inputs(): Validates runtime inputs
to_dict(): Includes parameters in serialization
Workflow.connect(): Validates compatible connections

get_performance_metrics() → list

Get collected performance metrics.

Return type:: list

get_security_context() → Dict[str, Any]

Get current security context.

Return type:: Dict[str, Any]

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

has_event_stream() → bool

Check if event stream is available.

Return type:: bool

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

set_event_context(event_stream: EventStream, session_id: str | None = None, workflow_id: str | None = None, execution_id: str | None = None)

Set the event context for this node.

Parameters:

event_stream (EventStream)
session_id (str | None)
workflow_id (str | None)
execution_id (str | None)

set_log_context(**context): Set logging context.

set_security_context(context: Dict[str, Any]) → None

Set security context for the node.

Parameters:: context (Dict[str, Any])
Return type:: None

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

Workflow export
Node persistence
API responses
Configuration sharing

The serialized format includes:

id: Unique node identifier
type: Node class name
metadata: Complete node metadata
config: Current configuration
parameters: Parameter definitions with types

Type serialization:

Python types are converted to string names
Complex types may require custom handling
Parameter defaults are included

Returns:

Node identification and type
Complete metadata
Configuration values
Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

WorkflowExporter: For workflow serialization
CLI: For node inspection
API: For node information endpoints
Debugging: For node state inspection

track_performance(func): Decorator to track method performance.

validate_inputs(**kwargs) → dict[str, Any]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

Gets parameter definitions from get_parameters()
Checks each parameter for:
- Presence (if required)
- Type compatibility
- Null handling for optional parameters
Attempts type conversion if needed
Applies default values for missing optional parameters

Key behaviors:

Required parameters must be provided or have defaults
Optional parameters can be None
Type mismatches attempt conversion before failing
Error messages include parameter descriptions

Example flow:: # Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

Type conversions applied
Defaults for missing optional parameters
None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

execute(): Before passing inputs to run()
Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

CycleAwareNode 

class kailash.nodes.base_cycle_aware.CycleAwareNode(**kwargs)[source]

Bases: Node

Base class for nodes that are cycle-aware with built-in helpers.

This class provides convenient methods for working with cyclic workflows, eliminating common boilerplate code for cycle information access and state management across iterations.

Design Philosophy:

CycleAwareNode is designed to make cyclic workflows as simple to write as regular nodes. It handles all the complexity of iteration tracking, state persistence, and cycle information management, allowing developers to focus on the iterative logic.

Upstream Dependencies:

Node: Base class that provides core node functionality
CyclicWorkflowExecutor: Provides cycle context in execution
Workflow: Must be configured with cycle=True connections

Downstream Consumers:

ConvergenceCheckerNode: Uses cycle helpers for convergence detection
A2ACoordinatorNode: Tracks agent performance across iterations
Any custom nodes needing cycle-aware behavior

Configuration:

No specific configuration required. Inherit from this class and use the provided helper methods in your run() implementation.

Implementation Details:

Extracts cycle information from execution context
Provides safe accessors with sensible defaults
Manages state persistence through _cycle_state convention
Offers utility methods for common patterns

Error Handling:

Returns default values if cycle information is missing
Handles missing state gracefully with empty dicts
Safe for use in non-cyclic contexts (acts as regular node)

Side Effects:

Logs cycle progress when log_cycle_info() is called
No other external side effects

Examples

>>> class QualityImproverNode(CycleAwareNode):
...     def run(self, **kwargs):
...         context = kwargs.get("context", {})
...         iteration = self.get_iteration(context)
...         quality = kwargs.get("quality", 0.0)
...
...         if self.is_first_iteration(context):
...             print("Starting quality improvement process")
...
...         # Improve quality based on iteration
...         improved_quality = quality + (0.1 * iteration)
...
...         return {
...             "quality": improved_quality,
...             **self.set_cycle_state({"last_quality": improved_quality})
...         }

get_cycle_info(context: dict[str, Any]) → dict[str, Any][source]

Get cycle information with sensible defaults.

Extracts cycle information from the execution context, providing default values for missing fields to prevent KeyError exceptions.

Parameters:

context (dict[str, Any]) – Execution context containing cycle information

Returns:

iteration: Current iteration number (default: 0)
elapsed_time: Time elapsed in seconds (default: 0.0)
cycle_id: Unique cycle identifier (default: “default”)
max_iterations: Maximum allowed iterations (default: 100)
start_time: Cycle start timestamp (default: current time)

Return type:

Dictionary containing cycle information with guaranteed fields

Example

>>> cycle_info = self.get_cycle_info(context)
>>> print(f"Iteration {cycle_info['iteration']} of {cycle_info['max_iterations']}")

get_iteration(context: dict[str, Any]) → int[source]

Get current iteration number.

Parameters:: context (dict[str, Any]) – Execution context
Returns:: Current iteration number (0-based)
Return type:: int

Example

>>> iteration = self.get_iteration(context)
>>> if iteration > 10:
...     print("Long-running cycle detected")

is_first_iteration(context: dict[str, Any]) → bool[source]

Check if this is the first iteration of the cycle.

Parameters:: context (dict[str, Any]) – Execution context
Returns:: True if this is iteration 0, False otherwise
Return type:: bool

Example

>>> if self.is_first_iteration(context):
...     print("Initializing cycle state")
...     return self.initialize_state()

is_last_iteration(context: dict[str, Any]) → bool[source]

Check if this is the last iteration of the cycle.

Parameters:: context (dict[str, Any]) – Execution context
Returns:: True if this is the final iteration, False otherwise
Return type:: bool

Example

>>> if self.is_last_iteration(context):
...     print("Performing final cleanup")

get_previous_state(context: dict[str, Any]) → dict[str, Any][source]

Get previous iteration state safely.

Retrieves state that was persisted from the previous iteration using set_cycle_state(). Returns empty dict if no state exists.

Parameters:: context (dict[str, Any]) – Execution context
Returns:: Dictionary containing state from previous iteration
Return type:: dict[str, Any]

Example

>>> prev_state = self.get_previous_state(context)
>>> history = prev_state.get("value_history", [])
>>> print(f"Previous values: {history}")

set_cycle_state(state: dict[str, Any]) → dict[str, Any][source]

Set state to persist to next iteration.

Creates the special _cycle_state return value that the cycle executor will persist and make available in the next iteration via get_previous_state().

Parameters:: state (dict[str, Any]) – Dictionary of state to persist
Returns:: Dictionary with _cycle_state key for return from run()
Return type:: dict[str, Any]

Example

>>> # In run() method:
>>> current_values = [1, 2, 3]
>>> return {
...     "result": processed_data,
...     **self.set_cycle_state({"values": current_values})
... }

get_cycle_progress(context: dict[str, Any]) → float[source]

Get cycle progress as a percentage.

Parameters:: context (dict[str, Any]) – Execution context
Returns:: Progress percentage (0.0 to 1.0)
Return type:: float

Example

>>> progress = self.get_cycle_progress(context)
>>> print(f"Cycle {progress*100:.1f}% complete")

log_cycle_info(context: dict[str, Any], message: str = '') → None[source]

Log cycle information for debugging.

Convenient method to log current cycle state with optional message.

Parameters:

context (dict[str, Any]) – Execution context
message (str) – Optional message to include in log

Return type:

None

Example

>>> self.log_cycle_info(context, "Processing batch")
# Output: [Cycle default] Iteration 3/10 (30.0%): Processing batch

should_continue_cycle(context: dict[str, Any], **kwargs) → bool[source]

Helper method to determine if cycle should continue.

This is a convenience method that can be overridden by subclasses to implement custom continuation logic. Default implementation checks if max iterations reached.

Parameters:

context (dict[str, Any]) – Execution context
**kwargs – Additional parameters for decision making

Returns:

True if cycle should continue, False otherwise

Return type:

bool

Example

>>> def should_continue_cycle(self, context, **kwargs):
...     quality = kwargs.get("quality", 0.0)
...     return quality < 0.95 and not self.is_last_iteration(context)

accumulate_values(context: dict[str, Any], key: str, value: Any, max_history: int = 100) → list[source]

Accumulate values across iterations with automatic history management.

Convenience method for maintaining a list of values across iterations with automatic size management to prevent memory growth.

Parameters:

context (dict[str, Any]) – Execution context
key (str) – State key for the value list
value (Any) – Value to add to the list
max_history (int) – Maximum number of values to keep

Returns:

Updated list of values

Return type:

list

Example

>>> quality_history = self.accumulate_values(context, "quality", current_quality)
>>> avg_quality = sum(quality_history) / len(quality_history)

detect_convergence_trend(context: dict[str, Any], key: str, threshold: float = 0.01, window: int = 3) → bool[source]

Detect if values are converging (becoming stable).

Analyzes recent values to determine if they are converging to a stable value.

Parameters:

context (dict[str, Any]) – Execution context
key (str) – State key containing value history
threshold (float) – Maximum variance for convergence
window (int) – Number of recent values to analyze

Returns:

True if values are converging, False otherwise

Return type:

bool

Example

>>> if self.detect_convergence_trend(context, "error_rate", 0.001):
...     return {"converged": True, "reason": "error_rate_stable"}

__init__(**kwargs)

Initialize the node with configuration parameters.

This method performs the following initialization steps:

Sets the node ID (defaults to class name)
Creates metadata from provided arguments
Sets up logging for the node
Stores configuration in self.config
Validates configuration against parameters

The configuration is validated by calling _validate_config(), which checks that all required parameters are present and of the correct type.

Parameters:: **kwargs – Configuration parameters including: - id: Optional custom node ID - name: Display name for the node - description: Node description - version: Node version - author: Node author - tags: Set of tags for discovery - Any parameters defined in get_parameters()
Raises:: NodeConfigurationError – If configuration is invalid or if metadata validation fails

Downstream effects:

Creates self.metadata for discovery
Sets up self.logger for execution logging
Stores self.config for runtime access
Validates parameters are correctly specified

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:: None

execute(**runtime_inputs) → dict[str, Any]

Execute the node with validation and error handling.

This is the main entry point for node execution that orchestrates the complete execution lifecycle:

Input validation (validate_inputs)
Execution (run)
Output validation (validate_outputs)
Error handling and logging
Performance metrics

Execution flow:

Logs execution start
Validates inputs against parameter schema
Calls run() with validated inputs
Validates outputs are JSON-serializable
Logs execution time
Returns validated outputs

Error handling strategy:

NodeValidationError: Re-raised as-is (input/output issues)
NodeExecutionError: Re-raised as-is (run() failures)
Other exceptions: Wrapped in NodeExecutionError

Performance tracking:

Records execution start/end times
Logs total execution duration
Includes timing in execution logs

Returns:

Dictionary of validated outputs from run()

Raises:

NodeExecutionError – If execution fails in run()
NodeValidationError – If input/output validation fails

Return type:

dict[str, Any]

Called by:

LocalRuntime: During workflow execution
TaskManager: With execution tracking
Unit tests: For node testing

Downstream effects:

Logs provide execution history
Metrics enable performance monitoring
Validation ensures data integrity

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_output_schema() → dict[str, NodeParameter]

Define output parameters for this node.

This optional method allows nodes to specify their output schema for validation. If not overridden, outputs will only be validated for JSON-serializability.

Design purpose: - Enables static analysis of node outputs - Provides runtime validation of output types - Supports automatic documentation of outputs - Facilitates workflow validation and type checking

The output schema serves similar purposes as input parameters:

Type validation during execution
Documentation for downstream consumers
Workflow connection validation
Export manifest generation

Example

>>> def get_output_schema(self):
...     return {
...         'dataframe': NodeParameter(
...             name='dataframe',
...             type=dict,
...             required=True,
...             description='Processed data as dictionary'
...         ),
...         'row_count': NodeParameter(
...             name='row_count',
...             type=int,
...             required=True,
...             description='Number of rows processed'
...         ),
...         'processing_time': NodeParameter(
...             name='processing_time',
...             type=float,
...             required=False,
...             description='Time taken to process in seconds'
...         )
...     }

Returns:: Dictionary mapping output names to their parameter definitions Empty dict by default (no schema validation)
Return type:: dict[str, NodeParameter]

Used by:

validate_outputs(): Validates runtime outputs
Workflow.connect(): Validates connections between nodes
Documentation generators: Create output documentation
Export systems: Include output schemas in manifests

abstractmethod get_parameters() → dict[str, NodeParameter]

Define the parameters this node accepts.

This abstract method must be implemented by all concrete nodes to specify their input schema. The parameters define:

What inputs the node expects
Type requirements for each input
Whether inputs are required or optional
Default values for optional inputs
Documentation for each parameter

The returned dictionary is used throughout the node lifecycle:

During initialization: _validate_config() checks configuration
During execution: validate_inputs() validates runtime data
During workflow creation: Used for connection validation
During export: Included in workflow manifests

Example

>>> def get_parameters(self):
...     return {
...         'input_file': NodeParameter(
...             name='input_file',
...             type=str,
...             required=True,
...             description='Path to input CSV file'
...         ),
...         'delimiter': NodeParameter(
...             name='delimiter',
...             type=str,
...             required=False,
...             default=',',
...             description='CSV delimiter character'
...         )
...     }

Returns:: Dictionary mapping parameter names to their definitions
Return type:: dict[str, NodeParameter]

Used by:

_validate_config(): Validates configuration matches parameters
validate_inputs(): Validates runtime inputs
to_dict(): Includes parameters in serialization
Workflow.connect(): Validates compatible connections

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

abstractmethod run(**kwargs) → dict[str, Any]

Execute the node’s logic.

This is the core method that implements the node’s data processing logic. It receives validated inputs and must return a dictionary of outputs.

Design requirements:

Must be stateless - no side effects between runs
All inputs are provided as keyword arguments
Must return a dictionary (JSON-serializable)
Should handle errors gracefully
Can use self.config for configuration values
Should use self.logger for status reporting

The method is called by execute() which handles:

Input validation before calling run()
Output validation after run() completes
Error wrapping and logging
Execution timing and metrics

Example

>>> def run(self, input_file, delimiter=','):
...     df = pd.read_csv(input_file, delimiter=delimiter)
...     return {
...         'dataframe': df.to_dict(),
...         'row_count': len(df),
...         'columns': list(df.columns)
...     }

Parameters:: **kwargs – Validated input parameters matching get_parameters()
Returns:: Dictionary of outputs that will be validated and passed to downstream nodes
Raises:: NodeExecutionError – If execution fails (will be caught and re-raised by execute())
Return type:: dict[str, Any]

Called by:

execute(): Wraps with validation and error handling
LocalRuntime: During workflow execution
TestRunner: During unit testing

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

Workflow export
Node persistence
API responses
Configuration sharing

The serialized format includes:

id: Unique node identifier
type: Node class name
metadata: Complete node metadata
config: Current configuration
parameters: Parameter definitions with types

Type serialization:

Python types are converted to string names
Complex types may require custom handling
Parameter defaults are included

Returns:

Node identification and type
Complete metadata
Configuration values
Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

WorkflowExporter: For workflow serialization
CLI: For node inspection
API: For node information endpoints
Debugging: For node state inspection

validate_inputs(**kwargs) → dict[str, Any]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

Gets parameter definitions from get_parameters()
Checks each parameter for:
- Presence (if required)
- Type compatibility
- Null handling for optional parameters
Attempts type conversion if needed
Applies default values for missing optional parameters

Key behaviors:

Required parameters must be provided or have defaults
Optional parameters can be None
Type mismatches attempt conversion before failing
Error messages include parameter descriptions

Example flow:: # Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

Type conversions applied
Defaults for missing optional parameters
None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

execute(): Before passing inputs to run()
Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

The CycleAwareNode provides built-in helpers for managing state and iteration tracking in cyclic workflows.

Helper Methods:

get_iteration(context): Get the current iteration number (0-based)
get_previous_state(context): Access state from the previous iteration
set_cycle_state(state): Persist state for the next iteration
accumulate_values(context, key, value): Build a rolling window of values
detect_convergence_trend(context, metric_key): Analyze convergence patterns
log_cycle_info(context, message): Log structured cycle information

Example Usage:

from kailash.nodes.base_cycle_aware import CycleAwareNode
from typing import Dict, Any

class OptimizerNode(CycleAwareNode):
    """Iterative optimization node that improves results each cycle."""

    def run(self, context: Dict[str, Any], **kwargs) -> Dict[str, Any]:
        # Get iteration info
        iteration = self.get_iteration(context)
        prev_state = self.get_previous_state(context)

        # Get previous value or start with initial
        current_value = prev_state.get("value", 0.5)

        # Optimization step
        improvement = 0.1 * (1 - current_value)  # Diminishing returns
        new_value = current_value + improvement

        # Track convergence
        self.accumulate_values(context, "value", new_value)
        trend = self.detect_convergence_trend(context, "value")

        # Save state for next iteration
        self.set_cycle_state({"value": new_value})

        # Log progress
        self.log_cycle_info(context, f"Value improved to {new_value:.3f}")

        # Check convergence
        converged = new_value > 0.95 or (trend["converging"] and trend["stability"] > 0.98)

        return {
            "value": new_value,
            "converged": converged,
            "iteration": iteration,
            "improvement": improvement
        }

# Use in a workflow
from kailash import Workflow

workflow = Workflow("optimization")
workflow.add_node("optimizer", OptimizerNode())

# Create optimization cycle
workflow.create_cycle("optimization_cycle") \
        .connect("optimizer", "optimizer", mapping={"value": "initial_value"}) \
        .max_iterations(100) \
        .converge_when("converged == True") \
        .build()

Common Patterns:

Retry with Exponential Backoff:

class RetryNode(CycleAwareNode):
    def run(self, context, **kwargs):
        iteration = self.get_iteration(context)
        max_retries = kwargs.get("max_retries", 3)

        try:
            result = perform_operation()
            return {"success": True, "result": result}
        except Exception as e:
            if iteration < max_retries:
                delay = 2 ** iteration  # Exponential backoff
                time.sleep(delay)
                self.log_cycle_info(context, f"Retry {iteration + 1} after {delay}s")
                return {"success": False, "retry": True}
            else:
                return {"success": False, "error": str(e)}

Data Quality Improvement:

class QualityImproverNode(CycleAwareNode):
    def run(self, context, **kwargs):
        data = kwargs["data"]
        prev_state = self.get_previous_state(context)

        # Calculate quality score
        quality = calculate_quality(data)
        self.accumulate_values(context, "quality", quality)

        # Check if we're making progress
        trend = self.detect_convergence_trend(context, "quality")
        if trend["plateau_detected"]:
            self.log_cycle_info(context, "Quality plateau detected")

        # Improve data
        improved_data = improve_quality(data)

        return {
            "data": improved_data,
            "quality": quality,
            "converged": quality > 0.95 or trend["plateau_detected"]
        }

Iterative Model Training:

class TrainingNode(CycleAwareNode):
    def run(self, context, **kwargs):
        model = kwargs.get("model")
        data = kwargs["training_data"]
        iteration = self.get_iteration(context)

        # Train for one epoch
        loss = model.train_epoch(data)
        self.accumulate_values(context, "loss", loss)

        # Early stopping check
        trend = self.detect_convergence_trend(context, "loss")
        early_stop = trend["converging"] and trend["stability"] > 0.99

        # Save best model
        prev_best = self.get_previous_state(context).get("best_loss", float('inf'))
        if loss < prev_best:
            self.set_cycle_state({"best_loss": loss, "best_model": model.state_dict()})

        return {
            "model": model,
            "loss": loss,
            "converged": early_stop or iteration >= 100
        }

Data Nodes 

Note

Additional data nodes are planned for future releases:

XMLReader/XMLWriter: For XML file processing
ParquetReader/ParquetWriter: For Apache Parquet columnar storage
ExcelReader/ExcelWriter: For Microsoft Excel files

Track implementation progress in the GitHub issues.

Data nodes handle input/output operations for various file formats and data sources.

DirectoryReaderNode (New in v0.2.1)

class kailash.nodes.data.directory.DirectoryReaderNode(**kwargs)[source]

Bases: Node

Discovers and catalogs files in a directory with metadata extraction.

This node provides comprehensive directory scanning capabilities, handling file discovery, metadata extraction, and filtering. It’s designed for batch file processing workflows and dynamic data source discovery.

Design Philosophy:

The DirectoryReaderNode embodies the principle of “dynamic data discovery.” Instead of hardcoding file paths, workflows can dynamically discover available data sources at runtime. This makes workflows more flexible and adaptable to changing data environments.

Features:

Recursive directory scanning
File type detection and filtering
Metadata extraction (size, timestamps, MIME types)
Pattern-based filtering
Security-validated path operations

Use Cases:

Batch file processing workflows
Dynamic data pipeline creation
File monitoring and cataloging
Multi-format document processing
Data lake exploration

Output Format:

Returns a structured catalog of discovered files with: - File paths and names - File types and MIME types - File sizes and timestamps - Directory structure information

get_parameters() → dict[str, NodeParameter][source]

Define input parameters for directory scanning.

Return type:: dict[str, NodeParameter]

run(**kwargs) → dict[str, Any][source]

Execute directory scanning operation.

Returns:

discovered_files: List of file information dictionaries
files_by_type: Files grouped by type
directory_stats: Summary statistics

Return type:

Dictionary containing

__init__(**kwargs)

Initialize the node with configuration parameters.

This method performs the following initialization steps:

Sets the node ID (defaults to class name)
Creates metadata from provided arguments
Sets up logging for the node
Stores configuration in self.config
Validates configuration against parameters

The configuration is validated by calling _validate_config(), which checks that all required parameters are present and of the correct type.

Parameters:: **kwargs – Configuration parameters including: - id: Optional custom node ID - name: Display name for the node - description: Node description - version: Node version - author: Node author - tags: Set of tags for discovery - Any parameters defined in get_parameters()
Raises:: NodeConfigurationError – If configuration is invalid or if metadata validation fails

Downstream effects:

Creates self.metadata for discovery
Sets up self.logger for execution logging
Stores self.config for runtime access
Validates parameters are correctly specified

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:: None

execute(**runtime_inputs) → dict[str, Any]

Execute the node with validation and error handling.

This is the main entry point for node execution that orchestrates the complete execution lifecycle:

Input validation (validate_inputs)
Execution (run)
Output validation (validate_outputs)
Error handling and logging
Performance metrics

Execution flow:

Logs execution start
Validates inputs against parameter schema
Calls run() with validated inputs
Validates outputs are JSON-serializable
Logs execution time
Returns validated outputs

Error handling strategy:

NodeValidationError: Re-raised as-is (input/output issues)
NodeExecutionError: Re-raised as-is (run() failures)
Other exceptions: Wrapped in NodeExecutionError

Performance tracking:

Records execution start/end times
Logs total execution duration
Includes timing in execution logs

Returns:

Dictionary of validated outputs from run()

Raises:

NodeExecutionError – If execution fails in run()
NodeValidationError – If input/output validation fails

Return type:

dict[str, Any]

Called by:

LocalRuntime: During workflow execution
TaskManager: With execution tracking
Unit tests: For node testing

Downstream effects:

Logs provide execution history
Metrics enable performance monitoring
Validation ensures data integrity

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_output_schema() → dict[str, NodeParameter]

Define output parameters for this node.

This optional method allows nodes to specify their output schema for validation. If not overridden, outputs will only be validated for JSON-serializability.

Design purpose: - Enables static analysis of node outputs - Provides runtime validation of output types - Supports automatic documentation of outputs - Facilitates workflow validation and type checking

The output schema serves similar purposes as input parameters:

Type validation during execution
Documentation for downstream consumers
Workflow connection validation
Export manifest generation

Example

>>> def get_output_schema(self):
...     return {
...         'dataframe': NodeParameter(
...             name='dataframe',
...             type=dict,
...             required=True,
...             description='Processed data as dictionary'
...         ),
...         'row_count': NodeParameter(
...             name='row_count',
...             type=int,
...             required=True,
...             description='Number of rows processed'
...         ),
...         'processing_time': NodeParameter(
...             name='processing_time',
...             type=float,
...             required=False,
...             description='Time taken to process in seconds'
...         )
...     }

Returns:: Dictionary mapping output names to their parameter definitions Empty dict by default (no schema validation)
Return type:: dict[str, NodeParameter]

Used by:

validate_outputs(): Validates runtime outputs
Workflow.connect(): Validates connections between nodes
Documentation generators: Create output documentation
Export systems: Include output schemas in manifests

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

Workflow export
Node persistence
API responses
Configuration sharing

The serialized format includes:

id: Unique node identifier
type: Node class name
metadata: Complete node metadata
config: Current configuration
parameters: Parameter definitions with types

Type serialization:

Python types are converted to string names
Complex types may require custom handling
Parameter defaults are included

Returns:

Node identification and type
Complete metadata
Configuration values
Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

WorkflowExporter: For workflow serialization
CLI: For node inspection
API: For node information endpoints
Debugging: For node state inspection

validate_inputs(**kwargs) → dict[str, Any]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

Gets parameter definitions from get_parameters()
Checks each parameter for:
- Presence (if required)
- Type compatibility
- Null handling for optional parameters
Attempts type conversion if needed
Applies default values for missing optional parameters

Key behaviors:

Required parameters must be provided or have defaults
Optional parameters can be None
Type mismatches attempt conversion before failing
Error messages include parameter descriptions

Example flow:: # Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

Type conversions applied
Defaults for missing optional parameters
None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

execute(): Before passing inputs to run()
Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

Example Usage:

from kailash.nodes.data import DirectoryReaderNode

# Discover files dynamically
dir_reader = DirectoryReaderNode(
    directory_path="./data/inputs",
    recursive=True,
    pattern="*.{csv,json,xml}",
    include_metadata=True
)

workflow.add_node("file_discoverer", dir_reader)

# Use different outputs for different purposes
workflow.connect(
    "file_discoverer", "csv_processor",
    mapping={"files_by_type": "files_by_type"}
)

workflow.connect(
    "file_discoverer", "stats_reporter",
    mapping={"directory_stats": "stats"}
)

Key Features:

Dynamic file discovery with pattern matching
MIME type detection and metadata extraction
Organized output by file type for typed processing
Performance optimization for large directories
Recursive scanning with configurable depth

Readers 

CSVReaderNode

class kailash.nodes.data.readers.CSVReaderNode(**kwargs)[source]

Bases: Node

Reads data from CSV files with automatic header detection and type inference.

This node provides comprehensive CSV file reading capabilities, handling various formats, encodings, and edge cases. It automatically detects headers, infers data types, and provides consistent structured output for downstream processing in Kailash workflows.

Design Philosophy:

The CSVReaderNode embodies the principle of “data accessibility without complexity.” It abstracts the intricacies of CSV parsing while providing flexibility for various formats. The design prioritizes memory efficiency, automatic format detection, and consistent output structure, making it easy to integrate diverse CSV data sources into workflows.

Upstream Dependencies:

File system providing CSV files
Workflow orchestrators specifying file paths
Configuration systems providing parsing options
Previous nodes generating CSV file paths
User inputs defining data sources

Downstream Consumers:

DataTransformNode: Processes tabular data
FilterNode: Applies row/column filtering
AggregatorNode: Summarizes data
PythonCodeNode: Custom data processing
WriterNodes: Exports to other formats
Visualization nodes: Creates charts
ML nodes: Uses as training data

Configuration:

The node supports extensive CSV parsing options: - Delimiter detection (comma, tab, pipe, etc.) - Header row identification - Encoding specification (UTF-8, Latin-1, etc.) - Quote character handling - Skip rows/comments functionality - Column type inference - Missing value handling

Implementation Details:

Uses Python’s csv module for robust parsing
Implements streaming for large files
Automatic delimiter detection when not specified
Header detection based on first row analysis
Type inference for numeric/date columns
Memory-efficient processing with generators
Unicode normalization for consistent encoding

Error Handling:

FileNotFoundError: Clear message with path
PermissionError: Access rights guidance
UnicodeDecodeError: Encoding detection hints
csv.Error: Malformed data diagnostics
EmptyFileError: Handles zero-byte files
Partial read recovery for corrupted files

Side Effects:

Reads from file system
May consume significant memory for large files
Creates file handles (properly closed)
Updates internal read statistics

Examples

>>> # Basic CSV reading with headers
>>> reader = CSVReaderNode()
>>> result = reader.execute(
...     file_path="customers.csv",
...     headers=True
... )
>>> assert isinstance(result["data"], list)
>>> assert all(isinstance(row, dict) for row in result["data"])
>>> # Example output:
>>> # result["data"] = [
>>> #     {"id": "1", "name": "John Doe", "age": "30"},
>>> #     {"id": "2", "name": "Jane Smith", "age": "25"}
>>> # ]
>>>
>>> # Reading with custom delimiter
>>> result = reader.execute(
...     file_path="data.tsv",
...     delimiter="\t",
...     headers=True
... )
>>>
>>> # Reading without headers (returns list of lists)
>>> result = reader.execute(
...     file_path="data.csv",
...     headers=False
... )
>>> assert all(isinstance(row, list) for row in result["data"])
>>>
>>> # Reading with specific encoding
>>> result = reader.execute(
...     file_path="european_data.csv",
...     encoding="iso-8859-1",
...     headers=True
... )
>>>
>>> # Handling quoted fields
>>> result = reader.execute(
...     file_path="complex.csv",
...     headers=True,
...     quotechar='"'
... )

get_parameters() → dict[str, NodeParameter][source]

Define input parameters for CSV reading.

This method specifies the configuration options for reading CSV files, providing flexibility while maintaining sensible defaults.

Parameter Design: 1. file_path: Required for locating the data source 2. headers: Optional with smart default (True) 3. delimiter: Optional with standard default (‘,’) 4. index_column: Optional column to use as dictionary key

The parameters are designed to handle common CSV variants while keeping the interface simple for typical use cases.

Returns:

Input validation during execution
UI generation for configuration
Workflow validation for connections
Documentation and help systems

Return type:

Dictionary of parameter definitions used by

run(**kwargs) → dict[str, Any][source]

Execute CSV reading operation.

This method performs the actual file reading, handling both headerless and header-based CSV formats. It uses Python’s csv module for robust parsing of various CSV dialects.

Processing Steps: 1. Opens file with UTF-8 encoding (standard) 2. Creates csv.reader with specified delimiter 3. Processes headers if present 4. Converts rows to appropriate format 5. Returns standardized output

Memory Considerations: - Loads entire file into memory - Suitable for files up to ~100MB - For larger files, consider streaming approach

Output Format: - With headers: List of dictionaries - Without headers: List of lists - With index_column: Also returns dictionary indexed by the column - Always wrapped in {“data”: …} for consistency

Parameters:

**kwargs – Validated parameters including: - file_path: Path to CSV file - headers: Whether to treat first row as headers - delimiter: Character separating values - index_column: Column to use as key for indexed dictionary

Returns:

‘data’ key containing list of dicts or lists
’data_indexed’ key (if index_column provided) containing dict

Return type:

Dictionary with

Raises:

FileNotFoundError – If file doesn’t exist
PermissionError – If file can’t be read
UnicodeDecodeError – If encoding is wrong
KeyError – If index_column doesn’t exist in headers

Downstream usage:

Transform nodes expect consistent data structure
Writers can directly output the data
Analyzers can process row-by-row
data_indexed is useful for lookups and joins

async async_run(**kwargs) → dict[str, Any][source]

Read CSV file asynchronously for better I/O performance.

This method provides true async file reading with aiofiles, offering significant performance improvements for large files and concurrent operations.

Parameters:: method (Same as run())
Returns:: Same as run() method
Raises:: Same as run() method –
Return type:: dict[str, Any]

__init__(**kwargs)

Initialize the node with configuration parameters.

This method performs the following initialization steps:

Sets the node ID (defaults to class name)
Creates metadata from provided arguments
Sets up logging for the node
Stores configuration in self.config
Validates configuration against parameters

The configuration is validated by calling _validate_config(), which checks that all required parameters are present and of the correct type.

Parameters:: **kwargs – Configuration parameters including: - id: Optional custom node ID - name: Display name for the node - description: Node description - version: Node version - author: Node author - tags: Set of tags for discovery - Any parameters defined in get_parameters()
Raises:: NodeConfigurationError – If configuration is invalid or if metadata validation fails

Downstream effects:

Creates self.metadata for discovery
Sets up self.logger for execution logging
Stores self.config for runtime access
Validates parameters are correctly specified

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:: None

execute(**runtime_inputs) → dict[str, Any]

Execute the node with validation and error handling.

This is the main entry point for node execution that orchestrates the complete execution lifecycle:

Input validation (validate_inputs)
Execution (run)
Output validation (validate_outputs)
Error handling and logging
Performance metrics

Execution flow:

Logs execution start
Validates inputs against parameter schema
Calls run() with validated inputs
Validates outputs are JSON-serializable
Logs execution time
Returns validated outputs

Error handling strategy:

NodeValidationError: Re-raised as-is (input/output issues)
NodeExecutionError: Re-raised as-is (run() failures)
Other exceptions: Wrapped in NodeExecutionError

Performance tracking:

Records execution start/end times
Logs total execution duration
Includes timing in execution logs

Returns:

Dictionary of validated outputs from run()

Raises:

NodeExecutionError – If execution fails in run()
NodeValidationError – If input/output validation fails

Return type:

dict[str, Any]

Called by:

LocalRuntime: During workflow execution
TaskManager: With execution tracking
Unit tests: For node testing

Downstream effects:

Logs provide execution history
Metrics enable performance monitoring
Validation ensures data integrity

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_output_schema() → dict[str, NodeParameter]

Define output parameters for this node.

This optional method allows nodes to specify their output schema for validation. If not overridden, outputs will only be validated for JSON-serializability.

Design purpose: - Enables static analysis of node outputs - Provides runtime validation of output types - Supports automatic documentation of outputs - Facilitates workflow validation and type checking

The output schema serves similar purposes as input parameters:

Type validation during execution
Documentation for downstream consumers
Workflow connection validation
Export manifest generation

Example

>>> def get_output_schema(self):
...     return {
...         'dataframe': NodeParameter(
...             name='dataframe',
...             type=dict,
...             required=True,
...             description='Processed data as dictionary'
...         ),
...         'row_count': NodeParameter(
...             name='row_count',
...             type=int,
...             required=True,
...             description='Number of rows processed'
...         ),
...         'processing_time': NodeParameter(
...             name='processing_time',
...             type=float,
...             required=False,
...             description='Time taken to process in seconds'
...         )
...     }

Returns:: Dictionary mapping output names to their parameter definitions Empty dict by default (no schema validation)
Return type:: dict[str, NodeParameter]

Used by:

validate_outputs(): Validates runtime outputs
Workflow.connect(): Validates connections between nodes
Documentation generators: Create output documentation
Export systems: Include output schemas in manifests

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

Workflow export
Node persistence
API responses
Configuration sharing

The serialized format includes:

id: Unique node identifier
type: Node class name
metadata: Complete node metadata
config: Current configuration
parameters: Parameter definitions with types

Type serialization:

Python types are converted to string names
Complex types may require custom handling
Parameter defaults are included

Returns:

Node identification and type
Complete metadata
Configuration values
Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

WorkflowExporter: For workflow serialization
CLI: For node inspection
API: For node information endpoints
Debugging: For node state inspection

validate_inputs(**kwargs) → dict[str, Any]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

Gets parameter definitions from get_parameters()
Checks each parameter for:
- Presence (if required)
- Type compatibility
- Null handling for optional parameters
Attempts type conversion if needed
Applies default values for missing optional parameters

Key behaviors:

Required parameters must be provided or have defaults
Optional parameters can be None
Type mismatches attempt conversion before failing
Error messages include parameter descriptions

Example flow:: # Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

Type conversions applied
Defaults for missing optional parameters
None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

execute(): Before passing inputs to run()
Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

Example Usage:

from kailash import Workflow
from kailash.nodes.data import CSVReaderNode

workflow = Workflow("csv_example")

# Create the CSV reader node
csv_reader = CSVReaderNode(
    file_path="customers.csv",
    encoding="utf-8"
)

# Add the node to the workflow
workflow.add_node("read_customers", csv_reader)

JSONReaderNode

class kailash.nodes.data.readers.JSONReaderNode(**kwargs)[source]

Bases: Node

Reads data from a JSON file.

This node handles JSON file reading with support for complex nested structures, arrays, and objects. It preserves the original JSON structure while ensuring compatibility with downstream nodes.

Design Features:

Preserves JSON structure integrity
Handles nested objects and arrays
Unicode-safe reading
Automatic type preservation
Memory-efficient for reasonable file sizes

Data Flow:

Input: JSON file path
Processing: Parse JSON maintaining structure
Output: Python objects matching JSON structure

Common Usage Patterns:

Loading configuration files
Reading API response caches
Processing structured data exports
Loading machine learning datasets

Upstream Sources:

API response saves
Configuration management
Data export systems
Previous JSONWriter outputs

Downstream Consumers:

Transform nodes: Process structured data
Logic nodes: Navigate JSON structure
JSONWriter: Re-export with modifications
AI nodes: Use as structured input

Error Handling:

FileNotFoundError: Missing file
json.JSONDecodeError: Invalid JSON syntax
PermissionError: Access denied
MemoryError: File too large

Example

# Read API response data reader = JSONReaderNode(file_path=’api_response.json’) result = reader.execute() # result[‘data’] = { # ‘status’: ‘success’, # ‘items’: [{‘id’: 1, ‘name’: ‘Item1’}], # ‘metadata’: {‘version’: ‘1.0’} # }

get_parameters() → dict[str, NodeParameter][source]

Define input parameters for JSON reading.

Simple parameter definition reflecting JSON’s self-describing nature. Unlike CSV, JSON files don’t require format configuration.

Design Choice: - Single required parameter for simplicity - No encoding parameter (UTF-8 standard for JSON) - No structure hints needed (self-describing format)

Returns:: Dictionary with single file_path parameter
Return type:: dict[str, NodeParameter]

run(**kwargs) → dict[str, Any][source]

Execute JSON reading operation.

Reads and parses JSON file, preserving the original structure and types. The json.load() function handles the parsing and type conversion automatically.

Processing Steps: 1. Opens file with UTF-8 encoding 2. Parses JSON to Python objects 3. Preserves structure (objects→dicts, arrays→lists) 4. Returns wrapped in standard format

Type Mappings: - JSON objects → Python dicts - JSON arrays → Python lists - JSON strings → Python strings - JSON numbers → Python int/float - JSON booleans → Python bool - JSON null → Python None

Parameters:

**kwargs – Validated parameters including: - file_path: Path to JSON file

Returns:

Dictionary with ‘data’ key containing the parsed JSON

Raises:

FileNotFoundError – If file doesn’t exist
json.JSONDecodeError – If JSON is malformed
PermissionError – If file can’t be read

Return type:

dict[str, Any]

Downstream usage:

Structure can be directly navigated
Compatible with JSONWriter for round-trip
Transform nodes can process nested data

async async_run(**kwargs) → dict[str, Any][source]

Read JSON file asynchronously for better I/O performance.

This method provides true async file reading with aiofiles, offering significant performance improvements for large files and concurrent operations.

Parameters:: method (Same as run())
Returns:: Same as run() method
Raises:: Same as run() method –
Return type:: dict[str, Any]

__init__(**kwargs)

Initialize the node with configuration parameters.

This method performs the following initialization steps:

Sets the node ID (defaults to class name)
Creates metadata from provided arguments
Sets up logging for the node
Stores configuration in self.config
Validates configuration against parameters

The configuration is validated by calling _validate_config(), which checks that all required parameters are present and of the correct type.

Parameters:: **kwargs – Configuration parameters including: - id: Optional custom node ID - name: Display name for the node - description: Node description - version: Node version - author: Node author - tags: Set of tags for discovery - Any parameters defined in get_parameters()
Raises:: NodeConfigurationError – If configuration is invalid or if metadata validation fails

Downstream effects:

Creates self.metadata for discovery
Sets up self.logger for execution logging
Stores self.config for runtime access
Validates parameters are correctly specified

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:: None

execute(**runtime_inputs) → dict[str, Any]

Execute the node with validation and error handling.

This is the main entry point for node execution that orchestrates the complete execution lifecycle:

Input validation (validate_inputs)
Execution (run)
Output validation (validate_outputs)
Error handling and logging
Performance metrics

Execution flow:

Logs execution start
Validates inputs against parameter schema
Calls run() with validated inputs
Validates outputs are JSON-serializable
Logs execution time
Returns validated outputs

Error handling strategy:

NodeValidationError: Re-raised as-is (input/output issues)
NodeExecutionError: Re-raised as-is (run() failures)
Other exceptions: Wrapped in NodeExecutionError

Performance tracking:

Records execution start/end times
Logs total execution duration
Includes timing in execution logs

Returns:

Dictionary of validated outputs from run()

Raises:

NodeExecutionError – If execution fails in run()
NodeValidationError – If input/output validation fails

Return type:

dict[str, Any]

Called by:

LocalRuntime: During workflow execution
TaskManager: With execution tracking
Unit tests: For node testing

Downstream effects:

Logs provide execution history
Metrics enable performance monitoring
Validation ensures data integrity

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_output_schema() → dict[str, NodeParameter]

Define output parameters for this node.

This optional method allows nodes to specify their output schema for validation. If not overridden, outputs will only be validated for JSON-serializability.

Design purpose: - Enables static analysis of node outputs - Provides runtime validation of output types - Supports automatic documentation of outputs - Facilitates workflow validation and type checking

The output schema serves similar purposes as input parameters:

Type validation during execution
Documentation for downstream consumers
Workflow connection validation
Export manifest generation

Example

>>> def get_output_schema(self):
...     return {
...         'dataframe': NodeParameter(
...             name='dataframe',
...             type=dict,
...             required=True,
...             description='Processed data as dictionary'
...         ),
...         'row_count': NodeParameter(
...             name='row_count',
...             type=int,
...             required=True,
...             description='Number of rows processed'
...         ),
...         'processing_time': NodeParameter(
...             name='processing_time',
...             type=float,
...             required=False,
...             description='Time taken to process in seconds'
...         )
...     }

Returns:: Dictionary mapping output names to their parameter definitions Empty dict by default (no schema validation)
Return type:: dict[str, NodeParameter]

Used by:

validate_outputs(): Validates runtime outputs
Workflow.connect(): Validates connections between nodes
Documentation generators: Create output documentation
Export systems: Include output schemas in manifests

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

Workflow export
Node persistence
API responses
Configuration sharing

The serialized format includes:

id: Unique node identifier
type: Node class name
metadata: Complete node metadata
config: Current configuration
parameters: Parameter definitions with types

Type serialization:

Python types are converted to string names
Complex types may require custom handling
Parameter defaults are included

Returns:

Node identification and type
Complete metadata
Configuration values
Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

WorkflowExporter: For workflow serialization
CLI: For node inspection
API: For node information endpoints
Debugging: For node state inspection

validate_inputs(**kwargs) → dict[str, Any]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

Gets parameter definitions from get_parameters()
Checks each parameter for:
- Presence (if required)
- Type compatibility
- Null handling for optional parameters
Attempts type conversion if needed
Applies default values for missing optional parameters

Key behaviors:

Required parameters must be provided or have defaults
Optional parameters can be None
Type mismatches attempt conversion before failing
Error messages include parameter descriptions

Example flow:: # Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

Type conversions applied
Defaults for missing optional parameters
None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

execute(): Before passing inputs to run()
Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

Example Usage:

workflow.add_node("JSONReaderNode", "read_config", config={
    "file_path": "config.json",
    "encoding": "utf-8"
})

TextReaderNode

class kailash.nodes.data.readers.TextReaderNode(**kwargs)[source]

Bases: Node

Reads text from a file.

This node provides simple text file reading with encoding support. It’s designed for processing plain text files, logs, documents, and any text-based format not handled by specialized readers.

Design Features: 1. Flexible encoding support 2. Reads entire file as single string 3. Preserves line endings and whitespace 4. Handles various text encodings 5. Simple, predictable output format

Data Flow: - Input: File path and encoding - Processing: Read entire file as text - Output: Single text string

Common Usage Patterns: 1. Reading log files 2. Processing documentation 3. Loading templates 4. Reading configuration files 5. Processing natural language data

Upstream Sources: - Log file generators - Document management systems - Template repositories - Previous TextWriter outputs

Downstream Consumers: - NLP processors: Analyze text content - Pattern matchers: Search for patterns - TextWriter: Save processed text - AI models: Process natural language

Error Handling: - FileNotFoundError: Missing file - PermissionError: Access denied - UnicodeDecodeError: Wrong encoding - MemoryError: File too large

Example

>>> # Read a log file
>>> reader = TextReaderNode(
...     file_path='application.log',
...     encoding='utf-8'
... )
>>> result = reader.execute()
>>> # result['text'] = "2024-01-01 INFO: Application started\n..."

get_parameters() → dict[str, NodeParameter][source]

Define input parameters for text reading.

Provides essential parameters for text file reading with encoding flexibility to handle international text.

Parameter Design: 1. file_path: Required for file location 2. encoding: Optional with UTF-8 default

The encoding parameter is crucial for: - International text support - Legacy system compatibility - Log file processing - Cross-platform text handling

Returns:: Dictionary of parameter definitions
Return type:: dict[str, NodeParameter]

run(**kwargs) → dict[str, Any][source]

Execute text reading operation.

Reads entire text file into memory as a single string, preserving all formatting, line endings, and whitespace.

Processing Steps: 1. Opens file with specified encoding 2. Reads entire content as string 3. Preserves original formatting 4. Returns in standard format

Memory Considerations: - Loads entire file into memory - Suitable for files up to ~10MB - Large files may need streaming approach

Output Note: - Returns {“text”: …} not {“data”: …} - Different from CSV/JSON readers for clarity - Text is unprocessed, raw content

Parameters:

**kwargs – Validated parameters including: - file_path: Path to text file - encoding: Character encoding

Returns:

Dictionary with ‘text’ key containing file content

Raises:

FileNotFoundError – If file doesn’t exist
UnicodeDecodeError – If encoding is incorrect
PermissionError – If file can’t be read

Return type:

dict[str, Any]

Downstream usage:

NLP nodes can tokenize/analyze
Pattern nodes can search content
Writers can save processed text

__init__(**kwargs)

Initialize the node with configuration parameters.

This method performs the following initialization steps:

Sets the node ID (defaults to class name)
Creates metadata from provided arguments
Sets up logging for the node
Stores configuration in self.config
Validates configuration against parameters

The configuration is validated by calling _validate_config(), which checks that all required parameters are present and of the correct type.

Parameters:: **kwargs – Configuration parameters including: - id: Optional custom node ID - name: Display name for the node - description: Node description - version: Node version - author: Node author - tags: Set of tags for discovery - Any parameters defined in get_parameters()
Raises:: NodeConfigurationError – If configuration is invalid or if metadata validation fails

Downstream effects:

Creates self.metadata for discovery
Sets up self.logger for execution logging
Stores self.config for runtime access
Validates parameters are correctly specified

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:: None

execute(**runtime_inputs) → dict[str, Any]

Execute the node with validation and error handling.

This is the main entry point for node execution that orchestrates the complete execution lifecycle:

Input validation (validate_inputs)
Execution (run)
Output validation (validate_outputs)
Error handling and logging
Performance metrics

Execution flow:

Logs execution start
Validates inputs against parameter schema
Calls run() with validated inputs
Validates outputs are JSON-serializable
Logs execution time
Returns validated outputs

Error handling strategy:

NodeValidationError: Re-raised as-is (input/output issues)
NodeExecutionError: Re-raised as-is (run() failures)
Other exceptions: Wrapped in NodeExecutionError

Performance tracking:

Records execution start/end times
Logs total execution duration
Includes timing in execution logs

Returns:

Dictionary of validated outputs from run()

Raises:

NodeExecutionError – If execution fails in run()
NodeValidationError – If input/output validation fails

Return type:

dict[str, Any]

Called by:

LocalRuntime: During workflow execution
TaskManager: With execution tracking
Unit tests: For node testing

Downstream effects:

Logs provide execution history
Metrics enable performance monitoring
Validation ensures data integrity

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_output_schema() → dict[str, NodeParameter]

Define output parameters for this node.

This optional method allows nodes to specify their output schema for validation. If not overridden, outputs will only be validated for JSON-serializability.

Design purpose: - Enables static analysis of node outputs - Provides runtime validation of output types - Supports automatic documentation of outputs - Facilitates workflow validation and type checking

The output schema serves similar purposes as input parameters:

Type validation during execution
Documentation for downstream consumers
Workflow connection validation
Export manifest generation

Example

>>> def get_output_schema(self):
...     return {
...         'dataframe': NodeParameter(
...             name='dataframe',
...             type=dict,
...             required=True,
...             description='Processed data as dictionary'
...         ),
...         'row_count': NodeParameter(
...             name='row_count',
...             type=int,
...             required=True,
...             description='Number of rows processed'
...         ),
...         'processing_time': NodeParameter(
...             name='processing_time',
...             type=float,
...             required=False,
...             description='Time taken to process in seconds'
...         )
...     }

Returns:: Dictionary mapping output names to their parameter definitions Empty dict by default (no schema validation)
Return type:: dict[str, NodeParameter]

Used by:

validate_outputs(): Validates runtime outputs
Workflow.connect(): Validates connections between nodes
Documentation generators: Create output documentation
Export systems: Include output schemas in manifests

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

Workflow export
Node persistence
API responses
Configuration sharing

The serialized format includes:

id: Unique node identifier
type: Node class name
metadata: Complete node metadata
config: Current configuration
parameters: Parameter definitions with types

Type serialization:

Python types are converted to string names
Complex types may require custom handling
Parameter defaults are included

Returns:

Node identification and type
Complete metadata
Configuration values
Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

WorkflowExporter: For workflow serialization
CLI: For node inspection
API: For node information endpoints
Debugging: For node state inspection

validate_inputs(**kwargs) → dict[str, Any]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

Gets parameter definitions from get_parameters()
Checks each parameter for:
- Presence (if required)
- Type compatibility
- Null handling for optional parameters
Attempts type conversion if needed
Applies default values for missing optional parameters

Key behaviors:

Required parameters must be provided or have defaults
Optional parameters can be None
Type mismatches attempt conversion before failing
Error messages include parameter descriptions

Example flow:: # Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

Type conversions applied
Defaults for missing optional parameters
None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

execute(): Before passing inputs to run()
Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

Writers 

CSVWriterNode

class kailash.nodes.data.writers.CSVWriterNode(**kwargs)[source]

Bases: Node

Writes data to a CSV file.

This node handles CSV file writing with support for both dictionary and list data structures. It automatically detects data format and applies appropriate writing strategies.

Design Features: 1. Automatic format detection (dict vs list) 2. Header generation from dictionary keys 3. Configurable delimiters 4. Unicode support through encoding 5. Transaction-safe writing

Data Flow: - Input: Structured data (list of dicts/lists) - Processing: Format detection and CSV generation - Output: File creation confirmation

Common Usage Patterns: 1. Exporting processed data 2. Creating reports 3. Generating data backups 4. Producing import files 5. Saving analysis results

Upstream Sources: - CSVReader: Modified data round-trip - Transform nodes: Processed tabular data - Aggregator: Summarized results - API nodes: Structured responses

Downstream Consumers: - File system: Stores the CSV - External tools: Excel, databases - Other workflows: Read the output - Archive systems: Long-term storage

Error Handling: - PermissionError: Write access denied - OSError: Disk full or path issues - TypeError: Invalid data structure - UnicodeEncodeError: Encoding issues

Example

>>> # Write customer data
>>> writer = CSVWriterNode(
...     file_path='output.csv',
...     data=[
...         {'id': 1, 'name': 'John', 'age': 30},
...         {'id': 2, 'name': 'Jane', 'age': 25}
...     ],
...     delimiter=','
... )
>>> result = writer.execute()
>>> # result = {'rows_written': 2, 'file_path': 'output.csv'}

get_parameters() → dict[str, NodeParameter][source]

Define input parameters for CSV writing.

Provides comprehensive parameters for flexible CSV output, supporting various data structures and formatting options.

Parameter Design: 1. file_path: Required output location 2. data: Required data to write 3. headers: Optional custom headers 4. delimiter: Optional separator

The parameters handle two main scenarios: - Dict data: Auto-extracts headers from keys - List data: Requires headers or writes raw

Returns:: Dictionary of parameter definitions for validation
Return type:: dict[str, NodeParameter]

run(**kwargs) → dict[str, Any][source]

Execute CSV writing operation.

Intelligently handles different data structures, automatically detecting format and applying appropriate writing strategy.

Processing Steps: 1. Detects data structure (dict vs list) 2. Determines headers (provided or extracted) 3. Creates appropriate CSV writer 4. Writes headers if applicable 5. Writes data rows 6. Returns write statistics

Format Detection: - Dict data: Uses DictWriter, auto-extracts headers - List data: Uses standard writer, optional headers - Empty data: Returns zero rows written

File Handling: - Creates new file (overwrites existing) - Uses UTF-8 encoding - Handles newlines correctly (cross-platform) - Closes file automatically

Parameters:

**kwargs – Validated parameters including: - file_path: Output file location - data: List of dicts or lists - headers: Optional column names - delimiter: Field separator

Returns:

rows_written: Number of data rows
file_path: Output file location

Return type:

Dictionary with

Raises:

PermissionError – If can’t write to location
OSError – If path issues occur
TypeError – If data format invalid

Downstream usage:

File can be read by CSVReader
External tools can process output
Metrics available for monitoring

__init__(**kwargs)

Initialize the node with configuration parameters.

This method performs the following initialization steps:

Sets the node ID (defaults to class name)
Creates metadata from provided arguments
Sets up logging for the node
Stores configuration in self.config
Validates configuration against parameters

The configuration is validated by calling _validate_config(), which checks that all required parameters are present and of the correct type.

Parameters:: **kwargs – Configuration parameters including: - id: Optional custom node ID - name: Display name for the node - description: Node description - version: Node version - author: Node author - tags: Set of tags for discovery - Any parameters defined in get_parameters()
Raises:: NodeConfigurationError – If configuration is invalid or if metadata validation fails

Downstream effects:

Creates self.metadata for discovery
Sets up self.logger for execution logging
Stores self.config for runtime access
Validates parameters are correctly specified

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:: None

execute(**runtime_inputs) → dict[str, Any]

Execute the node with validation and error handling.

This is the main entry point for node execution that orchestrates the complete execution lifecycle:

Input validation (validate_inputs)
Execution (run)
Output validation (validate_outputs)
Error handling and logging
Performance metrics

Execution flow:

Logs execution start
Validates inputs against parameter schema
Calls run() with validated inputs
Validates outputs are JSON-serializable
Logs execution time
Returns validated outputs

Error handling strategy:

NodeValidationError: Re-raised as-is (input/output issues)
NodeExecutionError: Re-raised as-is (run() failures)
Other exceptions: Wrapped in NodeExecutionError

Performance tracking:

Records execution start/end times
Logs total execution duration
Includes timing in execution logs

Returns:

Dictionary of validated outputs from run()

Raises:

NodeExecutionError – If execution fails in run()
NodeValidationError – If input/output validation fails

Return type:

dict[str, Any]

Called by:

LocalRuntime: During workflow execution
TaskManager: With execution tracking
Unit tests: For node testing

Downstream effects:

Logs provide execution history
Metrics enable performance monitoring
Validation ensures data integrity

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_output_schema() → dict[str, NodeParameter]

Define output parameters for this node.

This optional method allows nodes to specify their output schema for validation. If not overridden, outputs will only be validated for JSON-serializability.

Design purpose: - Enables static analysis of node outputs - Provides runtime validation of output types - Supports automatic documentation of outputs - Facilitates workflow validation and type checking

The output schema serves similar purposes as input parameters:

Type validation during execution
Documentation for downstream consumers
Workflow connection validation
Export manifest generation

Example

>>> def get_output_schema(self):
...     return {
...         'dataframe': NodeParameter(
...             name='dataframe',
...             type=dict,
...             required=True,
...             description='Processed data as dictionary'
...         ),
...         'row_count': NodeParameter(
...             name='row_count',
...             type=int,
...             required=True,
...             description='Number of rows processed'
...         ),
...         'processing_time': NodeParameter(
...             name='processing_time',
...             type=float,
...             required=False,
...             description='Time taken to process in seconds'
...         )
...     }

Returns:: Dictionary mapping output names to their parameter definitions Empty dict by default (no schema validation)
Return type:: dict[str, NodeParameter]

Used by:

validate_outputs(): Validates runtime outputs
Workflow.connect(): Validates connections between nodes
Documentation generators: Create output documentation
Export systems: Include output schemas in manifests

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

Workflow export
Node persistence
API responses
Configuration sharing

The serialized format includes:

id: Unique node identifier
type: Node class name
metadata: Complete node metadata
config: Current configuration
parameters: Parameter definitions with types

Type serialization:

Python types are converted to string names
Complex types may require custom handling
Parameter defaults are included

Returns:

Node identification and type
Complete metadata
Configuration values
Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

WorkflowExporter: For workflow serialization
CLI: For node inspection
API: For node information endpoints
Debugging: For node state inspection

validate_inputs(**kwargs) → dict[str, Any]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

Gets parameter definitions from get_parameters()
Checks each parameter for:
- Presence (if required)
- Type compatibility
- Null handling for optional parameters
Attempts type conversion if needed
Applies default values for missing optional parameters

Key behaviors:

Required parameters must be provided or have defaults
Optional parameters can be None
Type mismatches attempt conversion before failing
Error messages include parameter descriptions

Example flow:: # Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

Type conversions applied
Defaults for missing optional parameters
None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

execute(): Before passing inputs to run()
Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

Example Usage:

workflow.add_node("CSVWriterNode", "save_results", config={
    "file_path": "output/results.csv",
    "index": False,
    "encoding": "utf-8"
})

JSONWriterNode

class kailash.nodes.data.writers.JSONWriterNode(**kwargs)[source]

Bases: Node

Writes data to a JSON file.

This node handles JSON serialization with support for complex nested structures, pretty printing, and various data types. It ensures data persistence while maintaining structure integrity.

Design Features: 1. Preserves complex data structures 2. Pretty printing with indentation 3. Unicode support by default 4. Type preservation for round-trips 5. Atomic write operations

Data Flow: - Input: Any JSON-serializable data - Processing: JSON serialization - Output: File creation confirmation

Common Usage Patterns: 1. Saving API responses 2. Persisting configuration 3. Caching structured data 4. Exporting analysis results 5. Creating data backups

Upstream Sources: - JSONReader: Modified data round-trip - API nodes: Response data - Transform nodes: Processed structures - Aggregator: Complex results

Downstream Consumers: - File system: Stores JSON file - JSONReader: Can reload data - APIs: Import the data - Version control: Track changes

Error Handling: - TypeError: Non-serializable data - PermissionError: Write access denied - OSError: Path or disk issues - JSONEncodeError: Encoding problems

Example

>>> # Write API response
>>> writer = JSONWriterNode(
...     file_path='response.json',
...     data={
...         'status': 'success',
...         'results': [1, 2, 3],
...         'metadata': {'version': '1.0'}
...     },
...     indent=2
... )
>>> result = writer.execute()
>>> # result = {'file_path': 'response.json'}

get_parameters() → dict[str, NodeParameter][source]

Define input parameters for JSON writing.

Minimal parameters reflecting JSON’s flexibility while providing formatting control through indentation.

Parameter Design: 1. file_path: Required output location 2. data: Required data (any serializable) 3. indent: Optional formatting control

The ‘Any’ type for data reflects JSON’s ability to handle various structures - validation happens at serialization time.

Returns:: Dictionary of parameter definitions
Return type:: dict[str, NodeParameter]

run(**kwargs) → dict[str, Any][source]

Execute JSON writing operation.

Serializes data to JSON format with proper formatting and encoding. Handles complex nested structures while maintaining readability through indentation.

Processing Steps: 1. Opens file for writing 2. Serializes data to JSON 3. Applies formatting options 4. Ensures Unicode preservation 5. Writes atomically

Serialization Features: - Pretty printing with indentation - Unicode characters preserved - Consistent key ordering - Null value handling - Number precision maintained

Parameters:

**kwargs – Validated parameters including: - file_path: Output file location - data: Data to serialize - indent: Spaces for indentation

Returns:

file_path: Written file location

Return type:

Dictionary with

Raises:

TypeError – If data not serializable
PermissionError – If write denied
OSError – If path issues

Downstream usage:

JSONReader can reload file
Version control can track
APIs can import data

__init__(**kwargs)

Initialize the node with configuration parameters.

This method performs the following initialization steps:

Sets the node ID (defaults to class name)
Creates metadata from provided arguments
Sets up logging for the node
Stores configuration in self.config
Validates configuration against parameters

The configuration is validated by calling _validate_config(), which checks that all required parameters are present and of the correct type.

Parameters:: **kwargs – Configuration parameters including: - id: Optional custom node ID - name: Display name for the node - description: Node description - version: Node version - author: Node author - tags: Set of tags for discovery - Any parameters defined in get_parameters()
Raises:: NodeConfigurationError – If configuration is invalid or if metadata validation fails

Downstream effects:

Creates self.metadata for discovery
Sets up self.logger for execution logging
Stores self.config for runtime access
Validates parameters are correctly specified

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:: None

execute(**runtime_inputs) → dict[str, Any]

Execute the node with validation and error handling.

This is the main entry point for node execution that orchestrates the complete execution lifecycle:

Input validation (validate_inputs)
Execution (run)
Output validation (validate_outputs)
Error handling and logging
Performance metrics

Execution flow:

Logs execution start
Validates inputs against parameter schema
Calls run() with validated inputs
Validates outputs are JSON-serializable
Logs execution time
Returns validated outputs

Error handling strategy:

NodeValidationError: Re-raised as-is (input/output issues)
NodeExecutionError: Re-raised as-is (run() failures)
Other exceptions: Wrapped in NodeExecutionError

Performance tracking:

Records execution start/end times
Logs total execution duration
Includes timing in execution logs

Returns:

Dictionary of validated outputs from run()

Raises:

NodeExecutionError – If execution fails in run()
NodeValidationError – If input/output validation fails

Return type:

dict[str, Any]

Called by:

LocalRuntime: During workflow execution
TaskManager: With execution tracking
Unit tests: For node testing

Downstream effects:

Logs provide execution history
Metrics enable performance monitoring
Validation ensures data integrity

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_output_schema() → dict[str, NodeParameter]

Define output parameters for this node.

This optional method allows nodes to specify their output schema for validation. If not overridden, outputs will only be validated for JSON-serializability.

Design purpose: - Enables static analysis of node outputs - Provides runtime validation of output types - Supports automatic documentation of outputs - Facilitates workflow validation and type checking

The output schema serves similar purposes as input parameters:

Type validation during execution
Documentation for downstream consumers
Workflow connection validation
Export manifest generation

Example

>>> def get_output_schema(self):
...     return {
...         'dataframe': NodeParameter(
...             name='dataframe',
...             type=dict,
...             required=True,
...             description='Processed data as dictionary'
...         ),
...         'row_count': NodeParameter(
...             name='row_count',
...             type=int,
...             required=True,
...             description='Number of rows processed'
...         ),
...         'processing_time': NodeParameter(
...             name='processing_time',
...             type=float,
...             required=False,
...             description='Time taken to process in seconds'
...         )
...     }

Returns:: Dictionary mapping output names to their parameter definitions Empty dict by default (no schema validation)
Return type:: dict[str, NodeParameter]

Used by:

validate_outputs(): Validates runtime outputs
Workflow.connect(): Validates connections between nodes
Documentation generators: Create output documentation
Export systems: Include output schemas in manifests

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

Workflow export
Node persistence
API responses
Configuration sharing

The serialized format includes:

id: Unique node identifier
type: Node class name
metadata: Complete node metadata
config: Current configuration
parameters: Parameter definitions with types

Type serialization:

Python types are converted to string names
Complex types may require custom handling
Parameter defaults are included

Returns:

Node identification and type
Complete metadata
Configuration values
Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

WorkflowExporter: For workflow serialization
CLI: For node inspection
API: For node information endpoints
Debugging: For node state inspection

validate_inputs(**kwargs) → dict[str, Any]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

Gets parameter definitions from get_parameters()
Checks each parameter for:
- Presence (if required)
- Type compatibility
- Null handling for optional parameters
Attempts type conversion if needed
Applies default values for missing optional parameters

Key behaviors:

Required parameters must be provided or have defaults
Optional parameters can be None
Type mismatches attempt conversion before failing
Error messages include parameter descriptions

Example flow:: # Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

Type conversions applied
Defaults for missing optional parameters
None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

execute(): Before passing inputs to run()
Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

TextWriterNode

class kailash.nodes.data.writers.TextWriterNode(**kwargs)[source]

Bases: Node

Writes text to a file.

This node provides flexible text file writing with support for various encodings and append operations. It handles plain text output for logs, documents, and generated content.

Design Features: 1. Flexible encoding support 2. Append mode for log files 3. Overwrite mode for fresh output 4. Byte counting for verification 5. Unicode-safe operations

Data Flow: - Input: Text string and configuration - Processing: Encode and write text - Output: Write confirmation

Common Usage Patterns: 1. Writing log entries 2. Saving generated content 3. Creating documentation 4. Exporting text reports 5. Building configuration files

Upstream Sources: - TextReader: Modified text round-trip - Transform nodes: Processed text - AI nodes: Generated content - Template nodes: Formatted output

Downstream Consumers: - File system: Stores text file - Log analyzers: Process logs - Documentation systems: Use output - Version control: Track changes

Error Handling: - PermissionError: Write access denied - OSError: Path or disk issues - UnicodeEncodeError: Encoding mismatch - MemoryError: Text too large

Example

>>> # Append to log file
>>> writer = TextWriterNode(
...     file_path='app.log',
...     text='ERROR: Connection failed\n',
...     encoding='utf-8',
...     append=True
... )
>>> result = writer.execute()
>>> # result = {'file_path': 'app.log', 'bytes_written': 25}

get_parameters() → dict[str, NodeParameter][source]

Define input parameters for text writing.

Comprehensive parameters supporting various text writing scenarios from simple output to complex log management.

Parameter Design: 1. file_path: Required output location 2. text: Required content to write 3. encoding: Optional for compatibility 4. append: Optional for log patterns

The append parameter is crucial for: - Log file management - Continuous output streams - Building files incrementally - Preserving existing content

Returns:: Dictionary of parameter definitions
Return type:: dict[str, NodeParameter]

run(**kwargs) → dict[str, Any][source]

Execute text writing operation.

Writes text to file with specified encoding and mode. Supports both overwrite and append operations for different use cases like logging and content generation.

Processing Steps: 1. Determines write mode (append/overwrite) 2. Opens file with encoding 3. Writes text content 4. Calculates bytes written 5. Returns write statistics

Mode Selection: - append=False: Creates new or overwrites - append=True: Adds to existing file - File created if doesn’t exist (both modes)

Encoding Handling: - Encodes text before counting bytes - Supports any Python encoding - UTF-8 default for compatibility

Parameters:

**kwargs – Validated parameters including: - file_path: Output file location - text: Content to write - encoding: Character encoding - append: Write mode selection

Returns:

file_path: Written file location
bytes_written: Size of written data

Return type:

Dictionary with

Raises:

PermissionError – If write denied
OSError – If path issues
UnicodeEncodeError – If encoding fails

Downstream usage:

TextReader can read file
Log analyzers can process
Metrics available for monitoring

__init__(**kwargs)

Initialize the node with configuration parameters.

This method performs the following initialization steps:

Sets the node ID (defaults to class name)
Creates metadata from provided arguments
Sets up logging for the node
Stores configuration in self.config
Validates configuration against parameters

The configuration is validated by calling _validate_config(), which checks that all required parameters are present and of the correct type.

Parameters:: **kwargs – Configuration parameters including: - id: Optional custom node ID - name: Display name for the node - description: Node description - version: Node version - author: Node author - tags: Set of tags for discovery - Any parameters defined in get_parameters()
Raises:: NodeConfigurationError – If configuration is invalid or if metadata validation fails

Downstream effects:

Creates self.metadata for discovery
Sets up self.logger for execution logging
Stores self.config for runtime access
Validates parameters are correctly specified

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:: None

execute(**runtime_inputs) → dict[str, Any]

Execute the node with validation and error handling.

This is the main entry point for node execution that orchestrates the complete execution lifecycle:

Input validation (validate_inputs)
Execution (run)
Output validation (validate_outputs)
Error handling and logging
Performance metrics

Execution flow:

Logs execution start
Validates inputs against parameter schema
Calls run() with validated inputs
Validates outputs are JSON-serializable
Logs execution time
Returns validated outputs

Error handling strategy:

NodeValidationError: Re-raised as-is (input/output issues)
NodeExecutionError: Re-raised as-is (run() failures)
Other exceptions: Wrapped in NodeExecutionError

Performance tracking:

Records execution start/end times
Logs total execution duration
Includes timing in execution logs

Returns:

Dictionary of validated outputs from run()

Raises:

NodeExecutionError – If execution fails in run()
NodeValidationError – If input/output validation fails

Return type:

dict[str, Any]

Called by:

LocalRuntime: During workflow execution
TaskManager: With execution tracking
Unit tests: For node testing

Downstream effects:

Logs provide execution history
Metrics enable performance monitoring
Validation ensures data integrity

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_output_schema() → dict[str, NodeParameter]

Define output parameters for this node.

This optional method allows nodes to specify their output schema for validation. If not overridden, outputs will only be validated for JSON-serializability.

Design purpose: - Enables static analysis of node outputs - Provides runtime validation of output types - Supports automatic documentation of outputs - Facilitates workflow validation and type checking

The output schema serves similar purposes as input parameters:

Type validation during execution
Documentation for downstream consumers
Workflow connection validation
Export manifest generation

Example

>>> def get_output_schema(self):
...     return {
...         'dataframe': NodeParameter(
...             name='dataframe',
...             type=dict,
...             required=True,
...             description='Processed data as dictionary'
...         ),
...         'row_count': NodeParameter(
...             name='row_count',
...             type=int,
...             required=True,
...             description='Number of rows processed'
...         ),
...         'processing_time': NodeParameter(
...             name='processing_time',
...             type=float,
...             required=False,
...             description='Time taken to process in seconds'
...         )
...     }

Returns:: Dictionary mapping output names to their parameter definitions Empty dict by default (no schema validation)
Return type:: dict[str, NodeParameter]

Used by:

validate_outputs(): Validates runtime outputs
Workflow.connect(): Validates connections between nodes
Documentation generators: Create output documentation
Export systems: Include output schemas in manifests

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

Workflow export
Node persistence
API responses
Configuration sharing

The serialized format includes:

id: Unique node identifier
type: Node class name
metadata: Complete node metadata
config: Current configuration
parameters: Parameter definitions with types

Type serialization:

Python types are converted to string names
Complex types may require custom handling
Parameter defaults are included

Returns:

Node identification and type
Complete metadata
Configuration values
Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

WorkflowExporter: For workflow serialization
CLI: For node inspection
API: For node information endpoints
Debugging: For node state inspection

validate_inputs(**kwargs) → dict[str, Any]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

Gets parameter definitions from get_parameters()
Checks each parameter for:
- Presence (if required)
- Type compatibility
- Null handling for optional parameters
Attempts type conversion if needed
Applies default values for missing optional parameters

Key behaviors:

Required parameters must be provided or have defaults
Optional parameters can be None
Type mismatches attempt conversion before failing
Error messages include parameter descriptions

Example flow:: # Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

Type conversions applied
Defaults for missing optional parameters
None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

execute(): Before passing inputs to run()
Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

Database Nodes 

SQLDatabaseNode

class kailash.nodes.data.sql.SQLDatabaseNode(connection_string: str | None = None, pool_size: int = 5, max_overflow: int = 10, pool_timeout: int = 30, pool_recycle: int = 3600, pool_pre_ping: bool = True, echo: bool = False, connect_args: dict | None = None, **kwargs)[source]

Bases: Node

Parameters:

connection_string (str | None)
pool_size (int)
max_overflow (int)
pool_timeout (int)
pool_recycle (int)
pool_pre_ping (bool)
echo (bool)
connect_args (dict | None)

classmethod initialize(project_config_path: str) → None[source]

Initialize shared resources with project configuration.

DEPRECATED: Use direct configuration in constructor instead.

Parameters:: project_config_path (str) – Path to the project configuration YAML file
Return type:: None

__init__(connection_string: str | None = None, pool_size: int = 5, max_overflow: int = 10, pool_timeout: int = 30, pool_recycle: int = 3600, pool_pre_ping: bool = True, echo: bool = False, connect_args: dict | None = None, **kwargs)

Initialize SQLDatabaseNode with direct database connection configuration.

Parameters:

connection_string (str | None) – Database connection URL (e.g., “sqlite:///path/to/db.db”)
pool_size (int) – Number of connections in the pool (default: 5)
max_overflow (int) – Maximum overflow connections (default: 10)
pool_timeout (int) – Timeout in seconds to get connection from pool (default: 30)
pool_recycle (int) – Time in seconds to recycle connections (default: 3600)
pool_pre_ping (bool) – Test connections before use (default: True)
echo (bool) – Enable SQLAlchemy query logging (default: False)
connect_args (dict | None) – Additional database-specific connection arguments
**kwargs – Additional node configuration parameters

get_parameters() → dict[str, NodeParameter][source]

Define input parameters for SQL execution.

Configuration parameters (provided to constructor): 1. connection_string: Database connection URL 2. pool_size, max_overflow, etc.: Connection pool configuration

Runtime parameters (passed to run() method): 3. query: SQL query to execute 4. parameters: Query parameters for safety 5. result_format: Output format

Returns:: Dictionary of parameter definitions
Return type:: dict[str, NodeParameter]

run(**kwargs) → dict[str, Any][source]

Execute SQL query using shared connection pool.

Parameters:

**kwargs – Validated parameters including: - query: SQL statement - parameters: Query parameters (optional) - result_format: Output format (optional)

Returns:

data: Query results in specified format
row_count: Number of rows affected/returned
columns: List of column names
execution_time: Query execution duration

Return type:

Dictionary containing

Raises:

NodeExecutionError – Connection or query errors

async async_run(**kwargs) → dict[str, Any][source]

Async wrapper for the run method to maintain backward compatibility.

This method provides an async interface while maintaining the same functionality as the synchronous run method. The underlying SQLAlchemy operations are still synchronous but wrapped for async compatibility.

Parameters:: **kwargs – Same parameters as run()
Returns:: Same return format as run()
Return type:: dict[str, Any]

Note

This is a compatibility method. The actual database operations are still synchronous underneath.

classmethod get_pool_status() → dict[str, Any][source]

Get status of all shared connection pools.

Return type:: dict[str, Any]

classmethod cleanup_pools()[source]: Clean up all shared connection pools.

classmethod dispose_pools_for(connection_string: str) → int[source]

Dispose only the shared pools for a specific connection string.

Issue #1502: a bare-:memory: DataFlow owns a shared-cache SQLite DB keyed by a per-instance file:df_mem_<id>?mode=memory&cache=shared URI. The registry/state sync path acquires a StaticPool from _shared_pools (class-level) whose single connection keeps that shared-cache DB alive. DataFlow.close() MUST dispose that pool at teardown — otherwise the in-memory DB leaks for the process lifetime AND, because CPython reuses freed id() addresses, a later DataFlow(":memory:") at the same address computes the identical URI, hits the surviving pool, and aliases the prior instance’s data.

Targeted (matches cache_key[0] == connection_string) so it never touches other live instances’ pools — unlike the global cleanup_pools(). Returns the number of pools disposed.

Parameters:: connection_string (str)
Return type:: int

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:: None

execute(**runtime_inputs) → dict[str, Any]

Execute the node with validation and error handling.

This is the main entry point for node execution that orchestrates the complete execution lifecycle:

Input validation (validate_inputs)
Execution (run)
Output validation (validate_outputs)
Error handling and logging
Performance metrics

Execution flow:

Logs execution start
Validates inputs against parameter schema
Calls run() with validated inputs
Validates outputs are JSON-serializable
Logs execution time
Returns validated outputs

Error handling strategy:

NodeValidationError: Re-raised as-is (input/output issues)
NodeExecutionError: Re-raised as-is (run() failures)
Other exceptions: Wrapped in NodeExecutionError

Performance tracking:

Records execution start/end times
Logs total execution duration
Includes timing in execution logs

Returns:

Dictionary of validated outputs from run()

Raises:

NodeExecutionError – If execution fails in run()
NodeValidationError – If input/output validation fails

Return type:

dict[str, Any]

Called by:

LocalRuntime: During workflow execution
TaskManager: With execution tracking
Unit tests: For node testing

Downstream effects:

Logs provide execution history
Metrics enable performance monitoring
Validation ensures data integrity

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_output_schema() → dict[str, NodeParameter]

Define output parameters for this node.

This optional method allows nodes to specify their output schema for validation. If not overridden, outputs will only be validated for JSON-serializability.

Design purpose: - Enables static analysis of node outputs - Provides runtime validation of output types - Supports automatic documentation of outputs - Facilitates workflow validation and type checking

The output schema serves similar purposes as input parameters:

Type validation during execution
Documentation for downstream consumers
Workflow connection validation
Export manifest generation

Example

>>> def get_output_schema(self):
...     return {
...         'dataframe': NodeParameter(
...             name='dataframe',
...             type=dict,
...             required=True,
...             description='Processed data as dictionary'
...         ),
...         'row_count': NodeParameter(
...             name='row_count',
...             type=int,
...             required=True,
...             description='Number of rows processed'
...         ),
...         'processing_time': NodeParameter(
...             name='processing_time',
...             type=float,
...             required=False,
...             description='Time taken to process in seconds'
...         )
...     }

Returns:: Dictionary mapping output names to their parameter definitions Empty dict by default (no schema validation)
Return type:: dict[str, NodeParameter]

Used by:

validate_outputs(): Validates runtime outputs
Workflow.connect(): Validates connections between nodes
Documentation generators: Create output documentation
Export systems: Include output schemas in manifests

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

Workflow export
Node persistence
API responses
Configuration sharing

The serialized format includes:

id: Unique node identifier
type: Node class name
metadata: Complete node metadata
config: Current configuration
parameters: Parameter definitions with types

Type serialization:

Python types are converted to string names
Complex types may require custom handling
Parameter defaults are included

Returns:

Node identification and type
Complete metadata
Configuration values
Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

WorkflowExporter: For workflow serialization
CLI: For node inspection
API: For node information endpoints
Debugging: For node state inspection

validate_inputs(**kwargs) → dict[str, Any]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

Gets parameter definitions from get_parameters()
Checks each parameter for:
- Presence (if required)
- Type compatibility
- Null handling for optional parameters
Attempts type conversion if needed
Applies default values for missing optional parameters

Key behaviors:

Required parameters must be provided or have defaults
Optional parameters can be None
Type mismatches attempt conversion before failing
Error messages include parameter descriptions

Example flow:: # Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

Type conversions applied
Defaults for missing optional parameters
None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

execute(): Before passing inputs to run()
Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

Example Usage:

from kailash.nodes.data import SQLDatabaseNode

# Direct configuration approach (recommended)
db_node = SQLDatabaseNode(
    connection_string="sqlite:///data.db",
    pool_size=5,
    max_overflow=10
)

# Add to workflow
workflow.add_node("database", db_node)

# Execute with runtime parameters
result = db_node.run(
    query="SELECT * FROM customers WHERE active = ?",
    parameters=[True],
    result_format="dict"
)

# PostgreSQL with advanced configuration
pg_node = SQLDatabaseNode(
    connection_string="postgresql://user:pass@host/db",
    pool_size=10,
    max_overflow=20,
    pool_recycle=1800,
    connect_args={'connect_timeout': 10}
)

workflow.add_node("pg_database", pg_node)

# MySQL example
mysql_node = SQLDatabaseNode(
    connection_string="mysql+pymysql://user:pass@host/db",
    pool_size=8,
    echo=True  # Enable query logging
)

# Execute with different parameter styles
# SQLite uses ?
sqlite_result = db_node.run(
    query="SELECT * FROM users WHERE age > ? AND city = ?",
    parameters=[25, "New York"]
)

# PostgreSQL uses $1, $2, etc.
pg_result = pg_node.run(
    query="SELECT * FROM users WHERE age > $1 AND city = $2",
    parameters=[25, "New York"]
)

# MySQL uses %s
mysql_result = mysql_node.run(
    query="SELECT * FROM users WHERE age > %s AND city = %s",
    parameters=[25, "New York"]
)

Configuration Parameters:

connection_string (str, required): Database connection URL
- SQLite: sqlite:///path/to/database.db
- PostgreSQL: postgresql://user:password@host:port/database
- MySQL: mysql+pymysql://user:password@host:port/database
pool_size (int, optional): Number of connections in pool (default: 5)
max_overflow (int, optional): Maximum overflow connections (default: 10)
pool_timeout (int, optional): Timeout to get connection from pool (default: 30)
pool_recycle (int, optional): Time to recycle connections in seconds (default: 3600)
pool_pre_ping (bool, optional): Test connections before use (default: True)
echo (bool, optional): Enable SQLAlchemy query logging (default: False)
connect_args (dict, optional): Additional database-specific connection arguments

Runtime Parameters:

query (str, required): SQL query to execute
parameters (list, optional): Query parameters for safe execution
result_format (str, optional): Output format - ‘dict’, ‘list’, or ‘raw’ (default: ‘dict’)
timeout (int, optional): Query timeout in seconds

Security Features:

Parameterized queries prevent SQL injection
Connection string password masking in logs
Query safety validation warnings
Identifier sanitization for dynamic SQL
Error message sanitization

SharePoint Nodes 

SharePointGraphReader

class kailash.nodes.data.sharepoint_graph.SharePointGraphReader(**kwargs)[source]

Bases: Node

Node for reading files from SharePoint using Microsoft Graph API.

This node uses Microsoft Graph API with MSAL authentication, providing better compatibility with modern Azure AD app registrations compared to the legacy SharePoint REST API.

Key features: 1. Multiple authentication methods (client credentials, certificate, username/password, managed identity, device code) 2. Support for listing, downloading, and searching files 3. Folder navigation and library support 4. Stateless design for orchestration compatibility 5. JSON-serializable outputs for database persistence

Usage patterns: 1. List files in document libraries 2. Download files to local storage 3. Search for files by name 4. Navigate folder structures

Example (Client Credentials):

>>> reader = SharePointGraphReader()
>>> result = reader.execute(
...     auth_method="client_credentials",
...     tenant_id="your-tenant-id",
...     client_id="your-client-id",
...     client_secret="your-secret",
...     site_url="https://company.sharepoint.com/sites/project",
...     operation="list_files",
...     library_name="Documents",
...     folder_path="Reports/2024"
... )

Example (Certificate):

>>> reader = SharePointGraphReader()
>>> result = reader.execute(
...     auth_method="certificate",
...     tenant_id="your-tenant-id",
...     client_id="your-client-id",
...     certificate_path="/path/to/cert.pem",
...     site_url="https://company.sharepoint.com/sites/project",
...     operation="list_files"
... )

get_metadata() → NodeMetadata[source]

Get node metadata for discovery and orchestration.

Return type:: NodeMetadata

get_parameters() → dict[str, NodeParameter][source]

Define input parameters for SharePoint Graph operations with multiple auth methods.

Return type:: dict[str, NodeParameter]

run(**kwargs) → dict[str, Any][source]

Execute SharePoint Graph operation with selected authentication method.

This method is stateless and returns JSON-serializable results suitable for database persistence and orchestration.

Return type:: dict[str, Any]

__init__(**kwargs)

Initialize the node with configuration parameters.

This method performs the following initialization steps:

Sets the node ID (defaults to class name)
Creates metadata from provided arguments
Sets up logging for the node
Stores configuration in self.config
Validates configuration against parameters

The configuration is validated by calling _validate_config(), which checks that all required parameters are present and of the correct type.

Parameters:: **kwargs – Configuration parameters including: - id: Optional custom node ID - name: Display name for the node - description: Node description - version: Node version - author: Node author - tags: Set of tags for discovery - Any parameters defined in get_parameters()
Raises:: NodeConfigurationError – If configuration is invalid or if metadata validation fails

Downstream effects:

Creates self.metadata for discovery
Sets up self.logger for execution logging
Stores self.config for runtime access
Validates parameters are correctly specified

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:: None

execute(**runtime_inputs) → dict[str, Any]

Execute the node with validation and error handling.

This is the main entry point for node execution that orchestrates the complete execution lifecycle:

Input validation (validate_inputs)
Execution (run)
Output validation (validate_outputs)
Error handling and logging
Performance metrics

Execution flow:

Logs execution start
Validates inputs against parameter schema
Calls run() with validated inputs
Validates outputs are JSON-serializable
Logs execution time
Returns validated outputs

Error handling strategy:

NodeValidationError: Re-raised as-is (input/output issues)
NodeExecutionError: Re-raised as-is (run() failures)
Other exceptions: Wrapped in NodeExecutionError

Performance tracking:

Records execution start/end times
Logs total execution duration
Includes timing in execution logs

Returns:

Dictionary of validated outputs from run()

Raises:

NodeExecutionError – If execution fails in run()
NodeValidationError – If input/output validation fails

Return type:

dict[str, Any]

Called by:

LocalRuntime: During workflow execution
TaskManager: With execution tracking
Unit tests: For node testing

Downstream effects:

Logs provide execution history
Metrics enable performance monitoring
Validation ensures data integrity

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_output_schema() → dict[str, NodeParameter]

Define output parameters for this node.

This optional method allows nodes to specify their output schema for validation. If not overridden, outputs will only be validated for JSON-serializability.

Design purpose: - Enables static analysis of node outputs - Provides runtime validation of output types - Supports automatic documentation of outputs - Facilitates workflow validation and type checking

The output schema serves similar purposes as input parameters:

Type validation during execution
Documentation for downstream consumers
Workflow connection validation
Export manifest generation

Example

>>> def get_output_schema(self):
...     return {
...         'dataframe': NodeParameter(
...             name='dataframe',
...             type=dict,
...             required=True,
...             description='Processed data as dictionary'
...         ),
...         'row_count': NodeParameter(
...             name='row_count',
...             type=int,
...             required=True,
...             description='Number of rows processed'
...         ),
...         'processing_time': NodeParameter(
...             name='processing_time',
...             type=float,
...             required=False,
...             description='Time taken to process in seconds'
...         )
...     }

Returns:: Dictionary mapping output names to their parameter definitions Empty dict by default (no schema validation)
Return type:: dict[str, NodeParameter]

Used by:

validate_outputs(): Validates runtime outputs
Workflow.connect(): Validates connections between nodes
Documentation generators: Create output documentation
Export systems: Include output schemas in manifests

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

Workflow export
Node persistence
API responses
Configuration sharing

The serialized format includes:

id: Unique node identifier
type: Node class name
metadata: Complete node metadata
config: Current configuration
parameters: Parameter definitions with types

Type serialization:

Python types are converted to string names
Complex types may require custom handling
Parameter defaults are included

Returns:

Node identification and type
Complete metadata
Configuration values
Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

WorkflowExporter: For workflow serialization
CLI: For node inspection
API: For node information endpoints
Debugging: For node state inspection

validate_inputs(**kwargs) → dict[str, Any]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

Gets parameter definitions from get_parameters()
Checks each parameter for:
- Presence (if required)
- Type compatibility
- Null handling for optional parameters
Attempts type conversion if needed
Applies default values for missing optional parameters

Key behaviors:

Required parameters must be provided or have defaults
Optional parameters can be None
Type mismatches attempt conversion before failing
Error messages include parameter descriptions

Example flow:: # Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

Type conversions applied
Defaults for missing optional parameters
None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

execute(): Before passing inputs to run()
Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

Example Usage:

workflow.add_node("SharePointGraphReader", "read_docs", config={
    "tenant_id": "${TENANT_ID}",
    "client_id": "${CLIENT_ID}",
    "client_secret": "${CLIENT_SECRET}",
    "site_url": "https://company.sharepoint.com/sites/docs",
    "operation": "list_files",
    "library": "Shared Documents"
})

SharePointGraphWriter

class kailash.nodes.data.sharepoint_graph.SharePointGraphWriter(**kwargs)[source]

Bases: Node

Node for uploading files to SharePoint using Microsoft Graph API.

This node handles file uploads to SharePoint document libraries, supporting folder structures and metadata.

Example

>>> writer = SharePointGraphWriter()
>>> result = writer.execute(
...     tenant_id="your-tenant-id",
...     client_id="your-client-id",
...     client_secret="your-secret",
...     site_url="https://company.sharepoint.com/sites/project",
...     local_path="report.pdf",
...     library_name="Documents",
...     folder_path="Reports/2024",
...     sharepoint_name="Q4_Report_2024.pdf"
... )

get_metadata() → NodeMetadata[source]

Get node metadata for discovery and orchestration.

Return type:: NodeMetadata

get_parameters() → dict[str, NodeParameter][source]

Define input parameters for SharePoint upload operations.

Return type:: dict[str, NodeParameter]

run(**kwargs) → dict[str, Any][source]

Execute SharePoint upload operation.

Return type:: dict[str, Any]

__init__(**kwargs)

Initialize the node with configuration parameters.

This method performs the following initialization steps:

Sets the node ID (defaults to class name)
Creates metadata from provided arguments
Sets up logging for the node
Stores configuration in self.config
Validates configuration against parameters

The configuration is validated by calling _validate_config(), which checks that all required parameters are present and of the correct type.

Parameters:: **kwargs – Configuration parameters including: - id: Optional custom node ID - name: Display name for the node - description: Node description - version: Node version - author: Node author - tags: Set of tags for discovery - Any parameters defined in get_parameters()
Raises:: NodeConfigurationError – If configuration is invalid or if metadata validation fails

Downstream effects:

Creates self.metadata for discovery
Sets up self.logger for execution logging
Stores self.config for runtime access
Validates parameters are correctly specified

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:: None

execute(**runtime_inputs) → dict[str, Any]

Execute the node with validation and error handling.

This is the main entry point for node execution that orchestrates the complete execution lifecycle:

Input validation (validate_inputs)
Execution (run)
Output validation (validate_outputs)
Error handling and logging
Performance metrics

Execution flow:

Logs execution start
Validates inputs against parameter schema
Calls run() with validated inputs
Validates outputs are JSON-serializable
Logs execution time
Returns validated outputs

Error handling strategy:

NodeValidationError: Re-raised as-is (input/output issues)
NodeExecutionError: Re-raised as-is (run() failures)
Other exceptions: Wrapped in NodeExecutionError

Performance tracking:

Records execution start/end times
Logs total execution duration
Includes timing in execution logs

Returns:

Dictionary of validated outputs from run()

Raises:

NodeExecutionError – If execution fails in run()
NodeValidationError – If input/output validation fails

Return type:

dict[str, Any]

Called by:

LocalRuntime: During workflow execution
TaskManager: With execution tracking
Unit tests: For node testing

Downstream effects:

Logs provide execution history
Metrics enable performance monitoring
Validation ensures data integrity

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_output_schema() → dict[str, NodeParameter]

Define output parameters for this node.

This optional method allows nodes to specify their output schema for validation. If not overridden, outputs will only be validated for JSON-serializability.

Design purpose: - Enables static analysis of node outputs - Provides runtime validation of output types - Supports automatic documentation of outputs - Facilitates workflow validation and type checking

The output schema serves similar purposes as input parameters:

Type validation during execution
Documentation for downstream consumers
Workflow connection validation
Export manifest generation

Example

>>> def get_output_schema(self):
...     return {
...         'dataframe': NodeParameter(
...             name='dataframe',
...             type=dict,
...             required=True,
...             description='Processed data as dictionary'
...         ),
...         'row_count': NodeParameter(
...             name='row_count',
...             type=int,
...             required=True,
...             description='Number of rows processed'
...         ),
...         'processing_time': NodeParameter(
...             name='processing_time',
...             type=float,
...             required=False,
...             description='Time taken to process in seconds'
...         )
...     }

Returns:: Dictionary mapping output names to their parameter definitions Empty dict by default (no schema validation)
Return type:: dict[str, NodeParameter]

Used by:

validate_outputs(): Validates runtime outputs
Workflow.connect(): Validates connections between nodes
Documentation generators: Create output documentation
Export systems: Include output schemas in manifests

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

Workflow export
Node persistence
API responses
Configuration sharing

The serialized format includes:

id: Unique node identifier
type: Node class name
metadata: Complete node metadata
config: Current configuration
parameters: Parameter definitions with types

Type serialization:

Python types are converted to string names
Complex types may require custom handling
Parameter defaults are included

Returns:

Node identification and type
Complete metadata
Configuration values
Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

WorkflowExporter: For workflow serialization
CLI: For node inspection
API: For node information endpoints
Debugging: For node state inspection

validate_inputs(**kwargs) → dict[str, Any]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

Gets parameter definitions from get_parameters()
Checks each parameter for:
- Presence (if required)
- Type compatibility
- Null handling for optional parameters
Attempts type conversion if needed
Applies default values for missing optional parameters

Key behaviors:

Required parameters must be provided or have defaults
Optional parameters can be None
Type mismatches attempt conversion before failing
Error messages include parameter descriptions

Example flow:: # Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

Type conversions applied
Defaults for missing optional parameters
None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

execute(): Before passing inputs to run()
Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

Transform Nodes 

Transform nodes manipulate and process data.

FilterNode 

class kailash.nodes.transform.processors.FilterNode(**kwargs)[source]

Bases: Node

Filters data based on configurable conditions and operators.

This node provides flexible data filtering capabilities for lists and collections, supporting various comparison operators and field-based filtering for structured data. It’s designed to work seamlessly in data processing pipelines, reducing datasets to items that match specific criteria.

Design Philosophy:

The FilterNode embodies the principle of “declarative data selection.” Rather than writing custom filtering code, users declare their filtering criteria through simple configuration. The design supports both simple value filtering and complex field-based filtering for dictionaries, making it versatile for various data structures.

Upstream Dependencies:

Data source nodes providing lists to filter
Transform nodes producing structured data
Aggregation nodes generating collections
API nodes returning result sets
File readers loading datasets

Downstream Consumers:

Processing nodes working with filtered subsets
Aggregation nodes summarizing filtered data
Writer nodes exporting filtered results
Visualization nodes displaying subsets
Decision nodes based on filter results

Configuration:

The node supports flexible filtering options: - Field selection for dictionary filtering - Multiple comparison operators - Type-aware comparisons - Null value handling - String contains operations

Implementation Details:

Handles lists of any type (dicts, primitives, objects)
Type coercion for numeric comparisons
Null-safe operations
String conversion for contains operator
Preserves original data structure
Zero-copy filtering (returns references)

Error Handling:

Graceful handling of type mismatches
Null value comparison logic
Empty data returns empty result
Invalid field names return no matches
Operator errors fail safely

Side Effects:

No side effects (pure function)
Does not modify input data
Returns new filtered list

Examples

>>> # Filter list of numbers
>>> filter_node = FilterNode()
>>> result = filter_node.execute(
...     data=[1, 2, 3, 4, 5],
...     operator=">",
...     value=3
... )
>>> assert result["filtered_data"] == [4, 5]
>>>
>>> # Filter list of dictionaries by field
>>> users = [
...     {"name": "Alice", "age": 30},
...     {"name": "Bob", "age": 25},
...     {"name": "Charlie", "age": 35}
... ]
>>> result = filter_node.execute(
...     data=users,
...     field="age",
...     operator=">=",
...     value=30
... )
>>> assert len(result["filtered_data"]) == 2
>>> assert result["filtered_data"][0]["name"] == "Alice"
>>>
>>> # String contains filtering
>>> items = [
...     {"title": "Python Programming"},
...     {"title": "Java Development"},
...     {"title": "Python for Data Science"}
... ]
>>> result = filter_node.execute(
...     data=items,
...     field="title",
...     operator="contains",
...     value="Python"
... )
>>> assert len(result["filtered_data"]) == 2
>>>
>>> # Null value handling
>>> data_with_nulls = [
...     {"value": 10},
...     {"value": None},
...     {"value": 20}
... ]
>>> result = filter_node.execute(
...     data=data_with_nulls,
...     field="value",
...     operator="!=",
...     value=None
... )
>>> assert len(result["filtered_data"]) == 2

get_parameters() → dict[str, NodeParameter][source]

Define the parameters this node accepts.

This abstract method must be implemented by all concrete nodes to specify their input schema. The parameters define:

What inputs the node expects
Type requirements for each input
Whether inputs are required or optional
Default values for optional inputs
Documentation for each parameter

The returned dictionary is used throughout the node lifecycle:

During initialization: _validate_config() checks configuration
During execution: validate_inputs() validates runtime data
During workflow creation: Used for connection validation
During export: Included in workflow manifests

Example

>>> def get_parameters(self):
...     return {
...         'input_file': NodeParameter(
...             name='input_file',
...             type=str,
...             required=True,
...             description='Path to input CSV file'
...         ),
...         'delimiter': NodeParameter(
...             name='delimiter',
...             type=str,
...             required=False,
...             default=',',
...             description='CSV delimiter character'
...         )
...     }

Returns:: Dictionary mapping parameter names to their definitions
Return type:: dict[str, NodeParameter]

Used by:

_validate_config(): Validates configuration matches parameters
validate_inputs(): Validates runtime inputs
to_dict(): Includes parameters in serialization
Workflow.connect(): Validates compatible connections

run(**kwargs) → dict[str, Any][source]

Execute the node’s logic.

This is the core method that implements the node’s data processing logic. It receives validated inputs and must return a dictionary of outputs.

Design requirements:

Must be stateless - no side effects between runs
All inputs are provided as keyword arguments
Must return a dictionary (JSON-serializable)
Should handle errors gracefully
Can use self.config for configuration values
Should use self.logger for status reporting

The method is called by execute() which handles:

Input validation before calling run()
Output validation after run() completes
Error wrapping and logging
Execution timing and metrics

Example

>>> def run(self, input_file, delimiter=','):
...     df = pd.read_csv(input_file, delimiter=delimiter)
...     return {
...         'dataframe': df.to_dict(),
...         'row_count': len(df),
...         'columns': list(df.columns)
...     }

Parameters:: **kwargs – Validated input parameters matching get_parameters()
Returns:: Dictionary of outputs that will be validated and passed to downstream nodes
Raises:: NodeExecutionError – If execution fails (will be caught and re-raised by execute())
Return type:: dict[str, Any]

Called by:

execute(): Wraps with validation and error handling
LocalRuntime: During workflow execution
TestRunner: During unit testing

__init__(**kwargs)

Initialize the node with configuration parameters.

This method performs the following initialization steps:

Sets the node ID (defaults to class name)
Creates metadata from provided arguments
Sets up logging for the node
Stores configuration in self.config
Validates configuration against parameters

The configuration is validated by calling _validate_config(), which checks that all required parameters are present and of the correct type.

Parameters:: **kwargs – Configuration parameters including: - id: Optional custom node ID - name: Display name for the node - description: Node description - version: Node version - author: Node author - tags: Set of tags for discovery - Any parameters defined in get_parameters()
Raises:: NodeConfigurationError – If configuration is invalid or if metadata validation fails

Downstream effects:

Creates self.metadata for discovery
Sets up self.logger for execution logging
Stores self.config for runtime access
Validates parameters are correctly specified

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:: None

execute(**runtime_inputs) → dict[str, Any]

Execute the node with validation and error handling.

This is the main entry point for node execution that orchestrates the complete execution lifecycle:

Input validation (validate_inputs)
Execution (run)
Output validation (validate_outputs)
Error handling and logging
Performance metrics

Execution flow:

Logs execution start
Validates inputs against parameter schema
Calls run() with validated inputs
Validates outputs are JSON-serializable
Logs execution time
Returns validated outputs

Error handling strategy:

NodeValidationError: Re-raised as-is (input/output issues)
NodeExecutionError: Re-raised as-is (run() failures)
Other exceptions: Wrapped in NodeExecutionError

Performance tracking:

Records execution start/end times
Logs total execution duration
Includes timing in execution logs

Returns:

Dictionary of validated outputs from run()

Raises:

NodeExecutionError – If execution fails in run()
NodeValidationError – If input/output validation fails

Return type:

dict[str, Any]

Called by:

LocalRuntime: During workflow execution
TaskManager: With execution tracking
Unit tests: For node testing

Downstream effects:

Logs provide execution history
Metrics enable performance monitoring
Validation ensures data integrity

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_output_schema() → dict[str, NodeParameter]

Define output parameters for this node.

This optional method allows nodes to specify their output schema for validation. If not overridden, outputs will only be validated for JSON-serializability.

Design purpose: - Enables static analysis of node outputs - Provides runtime validation of output types - Supports automatic documentation of outputs - Facilitates workflow validation and type checking

The output schema serves similar purposes as input parameters:

Type validation during execution
Documentation for downstream consumers
Workflow connection validation
Export manifest generation

Example

>>> def get_output_schema(self):
...     return {
...         'dataframe': NodeParameter(
...             name='dataframe',
...             type=dict,
...             required=True,
...             description='Processed data as dictionary'
...         ),
...         'row_count': NodeParameter(
...             name='row_count',
...             type=int,
...             required=True,
...             description='Number of rows processed'
...         ),
...         'processing_time': NodeParameter(
...             name='processing_time',
...             type=float,
...             required=False,
...             description='Time taken to process in seconds'
...         )
...     }

Returns:: Dictionary mapping output names to their parameter definitions Empty dict by default (no schema validation)
Return type:: dict[str, NodeParameter]

Used by:

validate_outputs(): Validates runtime outputs
Workflow.connect(): Validates connections between nodes
Documentation generators: Create output documentation
Export systems: Include output schemas in manifests

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

Workflow export
Node persistence
API responses
Configuration sharing

The serialized format includes:

id: Unique node identifier
type: Node class name
metadata: Complete node metadata
config: Current configuration
parameters: Parameter definitions with types

Type serialization:

Python types are converted to string names
Complex types may require custom handling
Parameter defaults are included

Returns:

Node identification and type
Complete metadata
Configuration values
Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

WorkflowExporter: For workflow serialization
CLI: For node inspection
API: For node information endpoints
Debugging: For node state inspection

validate_inputs(**kwargs) → dict[str, Any]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

Gets parameter definitions from get_parameters()
Checks each parameter for:
- Presence (if required)
- Type compatibility
- Null handling for optional parameters
Attempts type conversion if needed
Applies default values for missing optional parameters

Key behaviors:

Required parameters must be provided or have defaults
Optional parameters can be None
Type mismatches attempt conversion before failing
Error messages include parameter descriptions

Example flow:: # Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

Type conversions applied
Defaults for missing optional parameters
None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

execute(): Before passing inputs to run()
Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

Example Usage:

from kailash.nodes.transform import FilterNode

# Filter numbers greater than a value
result = filter_node.run(
    data=[1, 2, 3, 4, 5],
    operator=">",
    value=3
)  # Returns: {"filtered_data": [4, 5]}

# Filter dictionaries by field
users = [
    {"name": "Alice", "age": 30},
    {"name": "Bob", "age": 25},
    {"name": "Charlie", "age": 35}
]
result = filter_node.run(
    data=users,
    field="age",
    operator=">=",
    value=30
)  # Returns users 30 and older

# String contains filtering
items = [
    {"title": "Python Programming"},
    {"title": "Java Development"},
    {"title": "Python for Data Science"}
]
result = filter_node.run(
    data=items,
    field="title",
    operator="contains",
    value="Python"
)  # Returns items with "Python" in title

Map 

class kailash.nodes.transform.processors.Map(**kwargs)[source]

Bases: Node

Maps data using a transformation.

get_parameters() → dict[str, NodeParameter][source]

Define the parameters this node accepts.

This abstract method must be implemented by all concrete nodes to specify their input schema. The parameters define:

What inputs the node expects
Type requirements for each input
Whether inputs are required or optional
Default values for optional inputs
Documentation for each parameter

The returned dictionary is used throughout the node lifecycle:

During initialization: _validate_config() checks configuration
During execution: validate_inputs() validates runtime data
During workflow creation: Used for connection validation
During export: Included in workflow manifests

Example

>>> def get_parameters(self):
...     return {
...         'input_file': NodeParameter(
...             name='input_file',
...             type=str,
...             required=True,
...             description='Path to input CSV file'
...         ),
...         'delimiter': NodeParameter(
...             name='delimiter',
...             type=str,
...             required=False,
...             default=',',
...             description='CSV delimiter character'
...         )
...     }

Returns:: Dictionary mapping parameter names to their definitions
Return type:: dict[str, NodeParameter]

Used by:

_validate_config(): Validates configuration matches parameters
validate_inputs(): Validates runtime inputs
to_dict(): Includes parameters in serialization
Workflow.connect(): Validates compatible connections

run(**kwargs) → dict[str, Any][source]

Execute the node’s logic.

This is the core method that implements the node’s data processing logic. It receives validated inputs and must return a dictionary of outputs.

Design requirements:

Must be stateless - no side effects between runs
All inputs are provided as keyword arguments
Must return a dictionary (JSON-serializable)
Should handle errors gracefully
Can use self.config for configuration values
Should use self.logger for status reporting

The method is called by execute() which handles:

Input validation before calling run()
Output validation after run() completes
Error wrapping and logging
Execution timing and metrics

Example

>>> def run(self, input_file, delimiter=','):
...     df = pd.read_csv(input_file, delimiter=delimiter)
...     return {
...         'dataframe': df.to_dict(),
...         'row_count': len(df),
...         'columns': list(df.columns)
...     }

Parameters:: **kwargs – Validated input parameters matching get_parameters()
Returns:: Dictionary of outputs that will be validated and passed to downstream nodes
Raises:: NodeExecutionError – If execution fails (will be caught and re-raised by execute())
Return type:: dict[str, Any]

Called by:

execute(): Wraps with validation and error handling
LocalRuntime: During workflow execution
TestRunner: During unit testing

__init__(**kwargs)

Initialize the node with configuration parameters.

This method performs the following initialization steps:

Sets the node ID (defaults to class name)
Creates metadata from provided arguments
Sets up logging for the node
Stores configuration in self.config
Validates configuration against parameters

The configuration is validated by calling _validate_config(), which checks that all required parameters are present and of the correct type.

Parameters:: **kwargs – Configuration parameters including: - id: Optional custom node ID - name: Display name for the node - description: Node description - version: Node version - author: Node author - tags: Set of tags for discovery - Any parameters defined in get_parameters()
Raises:: NodeConfigurationError – If configuration is invalid or if metadata validation fails

Downstream effects:

Creates self.metadata for discovery
Sets up self.logger for execution logging
Stores self.config for runtime access
Validates parameters are correctly specified

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:: None

execute(**runtime_inputs) → dict[str, Any]

Execute the node with validation and error handling.

This is the main entry point for node execution that orchestrates the complete execution lifecycle:

Input validation (validate_inputs)
Execution (run)
Output validation (validate_outputs)
Error handling and logging
Performance metrics

Execution flow:

Logs execution start
Validates inputs against parameter schema
Calls run() with validated inputs
Validates outputs are JSON-serializable
Logs execution time
Returns validated outputs

Error handling strategy:

NodeValidationError: Re-raised as-is (input/output issues)
NodeExecutionError: Re-raised as-is (run() failures)
Other exceptions: Wrapped in NodeExecutionError

Performance tracking:

Records execution start/end times
Logs total execution duration
Includes timing in execution logs

Returns:

Dictionary of validated outputs from run()

Raises:

NodeExecutionError – If execution fails in run()
NodeValidationError – If input/output validation fails

Return type:

dict[str, Any]

Called by:

LocalRuntime: During workflow execution
TaskManager: With execution tracking
Unit tests: For node testing

Downstream effects:

Logs provide execution history
Metrics enable performance monitoring
Validation ensures data integrity

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_output_schema() → dict[str, NodeParameter]

Define output parameters for this node.

This optional method allows nodes to specify their output schema for validation. If not overridden, outputs will only be validated for JSON-serializability.

Design purpose: - Enables static analysis of node outputs - Provides runtime validation of output types - Supports automatic documentation of outputs - Facilitates workflow validation and type checking

The output schema serves similar purposes as input parameters:

Type validation during execution
Documentation for downstream consumers
Workflow connection validation
Export manifest generation

Example

>>> def get_output_schema(self):
...     return {
...         'dataframe': NodeParameter(
...             name='dataframe',
...             type=dict,
...             required=True,
...             description='Processed data as dictionary'
...         ),
...         'row_count': NodeParameter(
...             name='row_count',
...             type=int,
...             required=True,
...             description='Number of rows processed'
...         ),
...         'processing_time': NodeParameter(
...             name='processing_time',
...             type=float,
...             required=False,
...             description='Time taken to process in seconds'
...         )
...     }

Returns:: Dictionary mapping output names to their parameter definitions Empty dict by default (no schema validation)
Return type:: dict[str, NodeParameter]

Used by:

validate_outputs(): Validates runtime outputs
Workflow.connect(): Validates connections between nodes
Documentation generators: Create output documentation
Export systems: Include output schemas in manifests

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

Workflow export
Node persistence
API responses
Configuration sharing

The serialized format includes:

id: Unique node identifier
type: Node class name
metadata: Complete node metadata
config: Current configuration
parameters: Parameter definitions with types

Type serialization:

Python types are converted to string names
Complex types may require custom handling
Parameter defaults are included

Returns:

Node identification and type
Complete metadata
Configuration values
Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

WorkflowExporter: For workflow serialization
CLI: For node inspection
API: For node information endpoints
Debugging: For node state inspection

validate_inputs(**kwargs) → dict[str, Any]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

Gets parameter definitions from get_parameters()
Checks each parameter for:
- Presence (if required)
- Type compatibility
- Null handling for optional parameters
Attempts type conversion if needed
Applies default values for missing optional parameters

Key behaviors:

Required parameters must be provided or have defaults
Optional parameters can be None
Type mismatches attempt conversion before failing
Error messages include parameter descriptions

Example flow:: # Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

Type conversions applied
Defaults for missing optional parameters
None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

execute(): Before passing inputs to run()
Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

Example Usage:

workflow.add_node("Map", "add_columns", config={
    "mapping": {
        "full_name": "lambda row: f'{row.first_name} {row.last_name}'",
        "is_vip": "lambda row: row.total_purchases > 10000",
        "category": "lambda row: 'Gold' if row.score > 80 else 'Silver'"
    }
})

Sort 

class kailash.nodes.transform.processors.Sort(**kwargs)[source]

Bases: Node

Sorts data.

get_parameters() → dict[str, NodeParameter][source]

Define the parameters this node accepts.

This abstract method must be implemented by all concrete nodes to specify their input schema. The parameters define:

What inputs the node expects
Type requirements for each input
Whether inputs are required or optional
Default values for optional inputs
Documentation for each parameter

The returned dictionary is used throughout the node lifecycle:

During initialization: _validate_config() checks configuration
During execution: validate_inputs() validates runtime data
During workflow creation: Used for connection validation
During export: Included in workflow manifests

Example

>>> def get_parameters(self):
...     return {
...         'input_file': NodeParameter(
...             name='input_file',
...             type=str,
...             required=True,
...             description='Path to input CSV file'
...         ),
...         'delimiter': NodeParameter(
...             name='delimiter',
...             type=str,
...             required=False,
...             default=',',
...             description='CSV delimiter character'
...         )
...     }

Returns:: Dictionary mapping parameter names to their definitions
Return type:: dict[str, NodeParameter]

Used by:

_validate_config(): Validates configuration matches parameters
validate_inputs(): Validates runtime inputs
to_dict(): Includes parameters in serialization
Workflow.connect(): Validates compatible connections

run(**kwargs) → dict[str, Any][source]

Execute the node’s logic.

This is the core method that implements the node’s data processing logic. It receives validated inputs and must return a dictionary of outputs.

Design requirements:

Must be stateless - no side effects between runs
All inputs are provided as keyword arguments
Must return a dictionary (JSON-serializable)
Should handle errors gracefully
Can use self.config for configuration values
Should use self.logger for status reporting

The method is called by execute() which handles:

Input validation before calling run()
Output validation after run() completes
Error wrapping and logging
Execution timing and metrics

Example

>>> def run(self, input_file, delimiter=','):
...     df = pd.read_csv(input_file, delimiter=delimiter)
...     return {
...         'dataframe': df.to_dict(),
...         'row_count': len(df),
...         'columns': list(df.columns)
...     }

Parameters:: **kwargs – Validated input parameters matching get_parameters()
Returns:: Dictionary of outputs that will be validated and passed to downstream nodes
Raises:: NodeExecutionError – If execution fails (will be caught and re-raised by execute())
Return type:: dict[str, Any]

Called by:

execute(): Wraps with validation and error handling
LocalRuntime: During workflow execution
TestRunner: During unit testing

__init__(**kwargs)

Initialize the node with configuration parameters.

This method performs the following initialization steps:

Sets the node ID (defaults to class name)
Creates metadata from provided arguments
Sets up logging for the node
Stores configuration in self.config
Validates configuration against parameters

The configuration is validated by calling _validate_config(), which checks that all required parameters are present and of the correct type.

Parameters:: **kwargs – Configuration parameters including: - id: Optional custom node ID - name: Display name for the node - description: Node description - version: Node version - author: Node author - tags: Set of tags for discovery - Any parameters defined in get_parameters()
Raises:: NodeConfigurationError – If configuration is invalid or if metadata validation fails

Downstream effects:

Creates self.metadata for discovery
Sets up self.logger for execution logging
Stores self.config for runtime access
Validates parameters are correctly specified

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:: None

execute(**runtime_inputs) → dict[str, Any]

Execute the node with validation and error handling.

This is the main entry point for node execution that orchestrates the complete execution lifecycle:

Input validation (validate_inputs)
Execution (run)
Output validation (validate_outputs)
Error handling and logging
Performance metrics

Execution flow:

Logs execution start
Validates inputs against parameter schema
Calls run() with validated inputs
Validates outputs are JSON-serializable
Logs execution time
Returns validated outputs

Error handling strategy:

NodeValidationError: Re-raised as-is (input/output issues)
NodeExecutionError: Re-raised as-is (run() failures)
Other exceptions: Wrapped in NodeExecutionError

Performance tracking:

Records execution start/end times
Logs total execution duration
Includes timing in execution logs

Returns:

Dictionary of validated outputs from run()

Raises:

NodeExecutionError – If execution fails in run()
NodeValidationError – If input/output validation fails

Return type:

dict[str, Any]

Called by:

LocalRuntime: During workflow execution
TaskManager: With execution tracking
Unit tests: For node testing

Downstream effects:

Logs provide execution history
Metrics enable performance monitoring
Validation ensures data integrity

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_output_schema() → dict[str, NodeParameter]

Define output parameters for this node.

This optional method allows nodes to specify their output schema for validation. If not overridden, outputs will only be validated for JSON-serializability.

Design purpose: - Enables static analysis of node outputs - Provides runtime validation of output types - Supports automatic documentation of outputs - Facilitates workflow validation and type checking

The output schema serves similar purposes as input parameters:

Type validation during execution
Documentation for downstream consumers
Workflow connection validation
Export manifest generation

Example

>>> def get_output_schema(self):
...     return {
...         'dataframe': NodeParameter(
...             name='dataframe',
...             type=dict,
...             required=True,
...             description='Processed data as dictionary'
...         ),
...         'row_count': NodeParameter(
...             name='row_count',
...             type=int,
...             required=True,
...             description='Number of rows processed'
...         ),
...         'processing_time': NodeParameter(
...             name='processing_time',
...             type=float,
...             required=False,
...             description='Time taken to process in seconds'
...         )
...     }

Returns:: Dictionary mapping output names to their parameter definitions Empty dict by default (no schema validation)
Return type:: dict[str, NodeParameter]

Used by:

validate_outputs(): Validates runtime outputs
Workflow.connect(): Validates connections between nodes
Documentation generators: Create output documentation
Export systems: Include output schemas in manifests

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

Workflow export
Node persistence
API responses
Configuration sharing

The serialized format includes:

id: Unique node identifier
type: Node class name
metadata: Complete node metadata
config: Current configuration
parameters: Parameter definitions with types

Type serialization:

Python types are converted to string names
Complex types may require custom handling
Parameter defaults are included

Returns:

Node identification and type
Complete metadata
Configuration values
Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

WorkflowExporter: For workflow serialization
CLI: For node inspection
API: For node information endpoints
Debugging: For node state inspection

validate_inputs(**kwargs) → dict[str, Any]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

Gets parameter definitions from get_parameters()
Checks each parameter for:
- Presence (if required)
- Type compatibility
- Null handling for optional parameters
Attempts type conversion if needed
Applies default values for missing optional parameters

Key behaviors:

Required parameters must be provided or have defaults
Optional parameters can be None
Type mismatches attempt conversion before failing
Error messages include parameter descriptions

Example flow:: # Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

Type conversions applied
Defaults for missing optional parameters
None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

execute(): Before passing inputs to run()
Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

Example Usage:

# Sort by field in ascending order
workflow.add_node("Sort", "sort_by_age", config={
    "field": "age",
    "reverse": False
})

# Sort by multiple criteria
workflow.add_node("Sort", "sort_complex", config={
    "field": "priority",
    "reverse": True  # Highest priority first
})

DataTransformer (Enhanced in v0.2.1)

class kailash.nodes.transform.processors.DataTransformer(**kwargs)[source]

Bases: Node

Transforms data using custom transformation functions provided as strings.

This node allows arbitrary data transformations by providing lambda functions or other Python code as strings. These are compiled and executed against the input data.

get_parameters() → dict[str, NodeParameter][source]

Define the parameters this node accepts.

This abstract method must be implemented by all concrete nodes to specify their input schema. The parameters define:

What inputs the node expects
Type requirements for each input
Whether inputs are required or optional
Default values for optional inputs
Documentation for each parameter

The returned dictionary is used throughout the node lifecycle:

During initialization: _validate_config() checks configuration
During execution: validate_inputs() validates runtime data
During workflow creation: Used for connection validation
During export: Included in workflow manifests

Example

>>> def get_parameters(self):
...     return {
...         'input_file': NodeParameter(
...             name='input_file',
...             type=str,
...             required=True,
...             description='Path to input CSV file'
...         ),
...         'delimiter': NodeParameter(
...             name='delimiter',
...             type=str,
...             required=False,
...             default=',',
...             description='CSV delimiter character'
...         )
...     }

Returns:: Dictionary mapping parameter names to their definitions
Return type:: dict[str, NodeParameter]

Used by:

_validate_config(): Validates configuration matches parameters
validate_inputs(): Validates runtime inputs
to_dict(): Includes parameters in serialization
Workflow.connect(): Validates compatible connections

validate_inputs(**kwargs) → dict[str, Any][source]

Override validate_inputs to accept arbitrary parameters for transformations.

DataTransformer needs to accept any input parameters that might be mapped from other nodes, not just the predefined parameters in get_parameters(). This enables flexible data flow in workflows.

Return type:: dict[str, Any]

run(**kwargs) → dict[str, Any][source]

Execute the node’s logic.

This is the core method that implements the node’s data processing logic. It receives validated inputs and must return a dictionary of outputs.

Design requirements:

Must be stateless - no side effects between runs
All inputs are provided as keyword arguments
Must return a dictionary (JSON-serializable)
Should handle errors gracefully
Can use self.config for configuration values
Should use self.logger for status reporting

The method is called by execute() which handles:

Input validation before calling run()
Output validation after run() completes
Error wrapping and logging
Execution timing and metrics

Example

>>> def run(self, input_file, delimiter=','):
...     df = pd.read_csv(input_file, delimiter=delimiter)
...     return {
...         'dataframe': df.to_dict(),
...         'row_count': len(df),
...         'columns': list(df.columns)
...     }

Parameters:: **kwargs – Validated input parameters matching get_parameters()
Returns:: Dictionary of outputs that will be validated and passed to downstream nodes
Raises:: NodeExecutionError – If execution fails (will be caught and re-raised by execute())
Return type:: dict[str, Any]

Called by:

execute(): Wraps with validation and error handling
LocalRuntime: During workflow execution
TestRunner: During unit testing

__init__(**kwargs)

Initialize the node with configuration parameters.

This method performs the following initialization steps:

Sets the node ID (defaults to class name)
Creates metadata from provided arguments
Sets up logging for the node
Stores configuration in self.config
Validates configuration against parameters

The configuration is validated by calling _validate_config(), which checks that all required parameters are present and of the correct type.

Parameters:: **kwargs – Configuration parameters including: - id: Optional custom node ID - name: Display name for the node - description: Node description - version: Node version - author: Node author - tags: Set of tags for discovery - Any parameters defined in get_parameters()
Raises:: NodeConfigurationError – If configuration is invalid or if metadata validation fails

Downstream effects:

Creates self.metadata for discovery
Sets up self.logger for execution logging
Stores self.config for runtime access
Validates parameters are correctly specified

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:: None

execute(**runtime_inputs) → dict[str, Any]

Execute the node with validation and error handling.

This is the main entry point for node execution that orchestrates the complete execution lifecycle:

Input validation (validate_inputs)
Execution (run)
Output validation (validate_outputs)
Error handling and logging
Performance metrics

Execution flow:

Logs execution start
Validates inputs against parameter schema
Calls run() with validated inputs
Validates outputs are JSON-serializable
Logs execution time
Returns validated outputs

Error handling strategy:

NodeValidationError: Re-raised as-is (input/output issues)
NodeExecutionError: Re-raised as-is (run() failures)
Other exceptions: Wrapped in NodeExecutionError

Performance tracking:

Records execution start/end times
Logs total execution duration
Includes timing in execution logs

Returns:

Dictionary of validated outputs from run()

Raises:

NodeExecutionError – If execution fails in run()
NodeValidationError – If input/output validation fails

Return type:

dict[str, Any]

Called by:

LocalRuntime: During workflow execution
TaskManager: With execution tracking
Unit tests: For node testing

Downstream effects:

Logs provide execution history
Metrics enable performance monitoring
Validation ensures data integrity

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_output_schema() → dict[str, NodeParameter]

Define output parameters for this node.

This optional method allows nodes to specify their output schema for validation. If not overridden, outputs will only be validated for JSON-serializability.

Design purpose: - Enables static analysis of node outputs - Provides runtime validation of output types - Supports automatic documentation of outputs - Facilitates workflow validation and type checking

The output schema serves similar purposes as input parameters:

Type validation during execution
Documentation for downstream consumers
Workflow connection validation
Export manifest generation

Example

>>> def get_output_schema(self):
...     return {
...         'dataframe': NodeParameter(
...             name='dataframe',
...             type=dict,
...             required=True,
...             description='Processed data as dictionary'
...         ),
...         'row_count': NodeParameter(
...             name='row_count',
...             type=int,
...             required=True,
...             description='Number of rows processed'
...         ),
...         'processing_time': NodeParameter(
...             name='processing_time',
...             type=float,
...             required=False,
...             description='Time taken to process in seconds'
...         )
...     }

Returns:: Dictionary mapping output names to their parameter definitions Empty dict by default (no schema validation)
Return type:: dict[str, NodeParameter]

Used by:

validate_outputs(): Validates runtime outputs
Workflow.connect(): Validates connections between nodes
Documentation generators: Create output documentation
Export systems: Include output schemas in manifests

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

Workflow export
Node persistence
API responses
Configuration sharing

The serialized format includes:

id: Unique node identifier
type: Node class name
metadata: Complete node metadata
config: Current configuration
parameters: Parameter definitions with types

Type serialization:

Python types are converted to string names
Complex types may require custom handling
Parameter defaults are included

Returns:

Node identification and type
Complete metadata
Configuration values
Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

WorkflowExporter: For workflow serialization
CLI: For node inspection
API: For node information endpoints
Debugging: For node state inspection

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

Example Usage:

workflow.add_node("DataTransformer", "transform", config={
    "operations": [
        {"type": "rename", "old": "cust_id", "new": "customer_id"},
        {"type": "cast", "column": "age", "dtype": "int"},
        {"type": "fillna", "column": "email", "value": "unknown@example.com"},
        {"type": "drop", "columns": ["temp_field", "debug_info"]}
    ]
})

Enhanced Parameter Mapping (v0.2.1):

DataTransformer now accepts arbitrary mapped parameters from other nodes, enabling more flexible data flow patterns:

# Connect with complex data mapping
workflow.connect(
    "file_discoverer", "processor",
    mapping={
        "files_by_type": "files_by_type",
        "directory_stats": "stats",
        "metadata": "file_metadata"
    }
)

processor = DataTransformer(transformations=['''
# All mapped parameters are now available
files_by_type = locals().get("files_by_type", {})
stats = locals().get("stats", {})
metadata = locals().get("file_metadata", {})

# Process the data
csv_files = files_by_type.get("csv", [])
result = {"processed_files": len(csv_files), "total_size": stats.get("total_size", 0)}
'''])

Bug Fixes in v0.2.1:

Fixed dictionary output bug where only keys were passed instead of full dictionaries
Enhanced input validation to accept arbitrary mapped parameters
Improved error handling and debugging capabilities

Text Processing 

HierarchicalChunkerNode

class kailash.nodes.transform.chunkers.HierarchicalChunkerNode(**kwargs)[source]

Bases: Node

Splits documents into hierarchical chunks for better retrieval.

get_parameters() → dict[str, NodeParameter][source]

Define the parameters this node accepts.

This abstract method must be implemented by all concrete nodes to specify their input schema. The parameters define:

What inputs the node expects
Type requirements for each input
Whether inputs are required or optional
Default values for optional inputs
Documentation for each parameter

The returned dictionary is used throughout the node lifecycle:

During initialization: _validate_config() checks configuration
During execution: validate_inputs() validates runtime data
During workflow creation: Used for connection validation
During export: Included in workflow manifests

Example

>>> def get_parameters(self):
...     return {
...         'input_file': NodeParameter(
...             name='input_file',
...             type=str,
...             required=True,
...             description='Path to input CSV file'
...         ),
...         'delimiter': NodeParameter(
...             name='delimiter',
...             type=str,
...             required=False,
...             default=',',
...             description='CSV delimiter character'
...         )
...     }

Returns:: Dictionary mapping parameter names to their definitions
Return type:: dict[str, NodeParameter]

Used by:

_validate_config(): Validates configuration matches parameters
validate_inputs(): Validates runtime inputs
to_dict(): Includes parameters in serialization
Workflow.connect(): Validates compatible connections

run(**kwargs) → dict[str, Any][source]

Execute the node’s logic.

This is the core method that implements the node’s data processing logic. It receives validated inputs and must return a dictionary of outputs.

Design requirements:

Must be stateless - no side effects between runs
All inputs are provided as keyword arguments
Must return a dictionary (JSON-serializable)
Should handle errors gracefully
Can use self.config for configuration values
Should use self.logger for status reporting

The method is called by execute() which handles:

Input validation before calling run()
Output validation after run() completes
Error wrapping and logging
Execution timing and metrics

Example

>>> def run(self, input_file, delimiter=','):
...     df = pd.read_csv(input_file, delimiter=delimiter)
...     return {
...         'dataframe': df.to_dict(),
...         'row_count': len(df),
...         'columns': list(df.columns)
...     }

Parameters:: **kwargs – Validated input parameters matching get_parameters()
Returns:: Dictionary of outputs that will be validated and passed to downstream nodes
Raises:: NodeExecutionError – If execution fails (will be caught and re-raised by execute())
Return type:: dict[str, Any]

Called by:

execute(): Wraps with validation and error handling
LocalRuntime: During workflow execution
TestRunner: During unit testing

__init__(**kwargs)

Initialize the node with configuration parameters.

This method performs the following initialization steps:

Sets the node ID (defaults to class name)
Creates metadata from provided arguments
Sets up logging for the node
Stores configuration in self.config
Validates configuration against parameters

The configuration is validated by calling _validate_config(), which checks that all required parameters are present and of the correct type.

Parameters:: **kwargs – Configuration parameters including: - id: Optional custom node ID - name: Display name for the node - description: Node description - version: Node version - author: Node author - tags: Set of tags for discovery - Any parameters defined in get_parameters()
Raises:: NodeConfigurationError – If configuration is invalid or if metadata validation fails

Downstream effects:

Creates self.metadata for discovery
Sets up self.logger for execution logging
Stores self.config for runtime access
Validates parameters are correctly specified

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:: None

execute(**runtime_inputs) → dict[str, Any]

Execute the node with validation and error handling.

This is the main entry point for node execution that orchestrates the complete execution lifecycle:

Input validation (validate_inputs)
Execution (run)
Output validation (validate_outputs)
Error handling and logging
Performance metrics

Execution flow:

Logs execution start
Validates inputs against parameter schema
Calls run() with validated inputs
Validates outputs are JSON-serializable
Logs execution time
Returns validated outputs

Error handling strategy:

NodeValidationError: Re-raised as-is (input/output issues)
NodeExecutionError: Re-raised as-is (run() failures)
Other exceptions: Wrapped in NodeExecutionError

Performance tracking:

Records execution start/end times
Logs total execution duration
Includes timing in execution logs

Returns:

Dictionary of validated outputs from run()

Raises:

NodeExecutionError – If execution fails in run()
NodeValidationError – If input/output validation fails

Return type:

dict[str, Any]

Called by:

LocalRuntime: During workflow execution
TaskManager: With execution tracking
Unit tests: For node testing

Downstream effects:

Logs provide execution history
Metrics enable performance monitoring
Validation ensures data integrity

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_output_schema() → dict[str, NodeParameter]

Define output parameters for this node.

This optional method allows nodes to specify their output schema for validation. If not overridden, outputs will only be validated for JSON-serializability.

Design purpose: - Enables static analysis of node outputs - Provides runtime validation of output types - Supports automatic documentation of outputs - Facilitates workflow validation and type checking

The output schema serves similar purposes as input parameters:

Type validation during execution
Documentation for downstream consumers
Workflow connection validation
Export manifest generation

Example

>>> def get_output_schema(self):
...     return {
...         'dataframe': NodeParameter(
...             name='dataframe',
...             type=dict,
...             required=True,
...             description='Processed data as dictionary'
...         ),
...         'row_count': NodeParameter(
...             name='row_count',
...             type=int,
...             required=True,
...             description='Number of rows processed'
...         ),
...         'processing_time': NodeParameter(
...             name='processing_time',
...             type=float,
...             required=False,
...             description='Time taken to process in seconds'
...         )
...     }

Returns:: Dictionary mapping output names to their parameter definitions Empty dict by default (no schema validation)
Return type:: dict[str, NodeParameter]

Used by:

validate_outputs(): Validates runtime outputs
Workflow.connect(): Validates connections between nodes
Documentation generators: Create output documentation
Export systems: Include output schemas in manifests

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

Workflow export
Node persistence
API responses
Configuration sharing

The serialized format includes:

id: Unique node identifier
type: Node class name
metadata: Complete node metadata
config: Current configuration
parameters: Parameter definitions with types

Type serialization:

Python types are converted to string names
Complex types may require custom handling
Parameter defaults are included

Returns:

Node identification and type
Complete metadata
Configuration values
Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

WorkflowExporter: For workflow serialization
CLI: For node inspection
API: For node information endpoints
Debugging: For node state inspection

validate_inputs(**kwargs) → dict[str, Any]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

Gets parameter definitions from get_parameters()
Checks each parameter for:
- Presence (if required)
- Type compatibility
- Null handling for optional parameters
Attempts type conversion if needed
Applies default values for missing optional parameters

Key behaviors:

Required parameters must be provided or have defaults
Optional parameters can be None
Type mismatches attempt conversion before failing
Error messages include parameter descriptions

Example flow:: # Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

Type conversions applied
Defaults for missing optional parameters
None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

execute(): Before passing inputs to run()
Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

ChunkTextExtractorNode

class kailash.nodes.transform.formatters.ChunkTextExtractorNode(**kwargs)[source]

Bases: Node

Extracts text content from chunks for embedding generation.

get_parameters() → dict[str, NodeParameter][source]

Define the parameters this node accepts.

This abstract method must be implemented by all concrete nodes to specify their input schema. The parameters define:

What inputs the node expects
Type requirements for each input
Whether inputs are required or optional
Default values for optional inputs
Documentation for each parameter

The returned dictionary is used throughout the node lifecycle:

During initialization: _validate_config() checks configuration
During execution: validate_inputs() validates runtime data
During workflow creation: Used for connection validation
During export: Included in workflow manifests

Example

>>> def get_parameters(self):
...     return {
...         'input_file': NodeParameter(
...             name='input_file',
...             type=str,
...             required=True,
...             description='Path to input CSV file'
...         ),
...         'delimiter': NodeParameter(
...             name='delimiter',
...             type=str,
...             required=False,
...             default=',',
...             description='CSV delimiter character'
...         )
...     }

Returns:: Dictionary mapping parameter names to their definitions
Return type:: dict[str, NodeParameter]

Used by:

_validate_config(): Validates configuration matches parameters
validate_inputs(): Validates runtime inputs
to_dict(): Includes parameters in serialization
Workflow.connect(): Validates compatible connections

run(**kwargs) → dict[str, Any][source]

Execute the node’s logic.

This is the core method that implements the node’s data processing logic. It receives validated inputs and must return a dictionary of outputs.

Design requirements:

Must be stateless - no side effects between runs
All inputs are provided as keyword arguments
Must return a dictionary (JSON-serializable)
Should handle errors gracefully
Can use self.config for configuration values
Should use self.logger for status reporting

The method is called by execute() which handles:

Input validation before calling run()
Output validation after run() completes
Error wrapping and logging
Execution timing and metrics

Example

>>> def run(self, input_file, delimiter=','):
...     df = pd.read_csv(input_file, delimiter=delimiter)
...     return {
...         'dataframe': df.to_dict(),
...         'row_count': len(df),
...         'columns': list(df.columns)
...     }

Parameters:: **kwargs – Validated input parameters matching get_parameters()
Returns:: Dictionary of outputs that will be validated and passed to downstream nodes
Raises:: NodeExecutionError – If execution fails (will be caught and re-raised by execute())
Return type:: dict[str, Any]

Called by:

execute(): Wraps with validation and error handling
LocalRuntime: During workflow execution
TestRunner: During unit testing

__init__(**kwargs)

Initialize the node with configuration parameters.

This method performs the following initialization steps:

Sets the node ID (defaults to class name)
Creates metadata from provided arguments
Sets up logging for the node
Stores configuration in self.config
Validates configuration against parameters

The configuration is validated by calling _validate_config(), which checks that all required parameters are present and of the correct type.

Parameters:: **kwargs – Configuration parameters including: - id: Optional custom node ID - name: Display name for the node - description: Node description - version: Node version - author: Node author - tags: Set of tags for discovery - Any parameters defined in get_parameters()
Raises:: NodeConfigurationError – If configuration is invalid or if metadata validation fails

Downstream effects:

Creates self.metadata for discovery
Sets up self.logger for execution logging
Stores self.config for runtime access
Validates parameters are correctly specified

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:: None

execute(**runtime_inputs) → dict[str, Any]

Execute the node with validation and error handling.

This is the main entry point for node execution that orchestrates the complete execution lifecycle:

Input validation (validate_inputs)
Execution (run)
Output validation (validate_outputs)
Error handling and logging
Performance metrics

Execution flow:

Logs execution start
Validates inputs against parameter schema
Calls run() with validated inputs
Validates outputs are JSON-serializable
Logs execution time
Returns validated outputs

Error handling strategy:

NodeValidationError: Re-raised as-is (input/output issues)
NodeExecutionError: Re-raised as-is (run() failures)
Other exceptions: Wrapped in NodeExecutionError

Performance tracking:

Records execution start/end times
Logs total execution duration
Includes timing in execution logs

Returns:

Dictionary of validated outputs from run()

Raises:

NodeExecutionError – If execution fails in run()
NodeValidationError – If input/output validation fails

Return type:

dict[str, Any]

Called by:

LocalRuntime: During workflow execution
TaskManager: With execution tracking
Unit tests: For node testing

Downstream effects:

Logs provide execution history
Metrics enable performance monitoring
Validation ensures data integrity

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_output_schema() → dict[str, NodeParameter]

Define output parameters for this node.

This optional method allows nodes to specify their output schema for validation. If not overridden, outputs will only be validated for JSON-serializability.

Design purpose: - Enables static analysis of node outputs - Provides runtime validation of output types - Supports automatic documentation of outputs - Facilitates workflow validation and type checking

The output schema serves similar purposes as input parameters:

Type validation during execution
Documentation for downstream consumers
Workflow connection validation
Export manifest generation

Example

>>> def get_output_schema(self):
...     return {
...         'dataframe': NodeParameter(
...             name='dataframe',
...             type=dict,
...             required=True,
...             description='Processed data as dictionary'
...         ),
...         'row_count': NodeParameter(
...             name='row_count',
...             type=int,
...             required=True,
...             description='Number of rows processed'
...         ),
...         'processing_time': NodeParameter(
...             name='processing_time',
...             type=float,
...             required=False,
...             description='Time taken to process in seconds'
...         )
...     }

Returns:: Dictionary mapping output names to their parameter definitions Empty dict by default (no schema validation)
Return type:: dict[str, NodeParameter]

Used by:

validate_outputs(): Validates runtime outputs
Workflow.connect(): Validates connections between nodes
Documentation generators: Create output documentation
Export systems: Include output schemas in manifests

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

Workflow export
Node persistence
API responses
Configuration sharing

The serialized format includes:

id: Unique node identifier
type: Node class name
metadata: Complete node metadata
config: Current configuration
parameters: Parameter definitions with types

Type serialization:

Python types are converted to string names
Complex types may require custom handling
Parameter defaults are included

Returns:

Node identification and type
Complete metadata
Configuration values
Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

WorkflowExporter: For workflow serialization
CLI: For node inspection
API: For node information endpoints
Debugging: For node state inspection

validate_inputs(**kwargs) → dict[str, Any]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

Gets parameter definitions from get_parameters()
Checks each parameter for:
- Presence (if required)
- Type compatibility
- Null handling for optional parameters
Attempts type conversion if needed
Applies default values for missing optional parameters

Key behaviors:

Required parameters must be provided or have defaults
Optional parameters can be None
Type mismatches attempt conversion before failing
Error messages include parameter descriptions

Example flow:: # Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

Type conversions applied
Defaults for missing optional parameters
None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

execute(): Before passing inputs to run()
Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

QueryTextWrapperNode

class kailash.nodes.transform.formatters.QueryTextWrapperNode(**kwargs)[source]

Bases: Node

Wraps query string in list for embedding generation.

get_parameters() → dict[str, NodeParameter][source]

Define the parameters this node accepts.

This abstract method must be implemented by all concrete nodes to specify their input schema. The parameters define:

What inputs the node expects
Type requirements for each input
Whether inputs are required or optional
Default values for optional inputs
Documentation for each parameter

The returned dictionary is used throughout the node lifecycle:

During initialization: _validate_config() checks configuration
During execution: validate_inputs() validates runtime data
During workflow creation: Used for connection validation
During export: Included in workflow manifests

Example

>>> def get_parameters(self):
...     return {
...         'input_file': NodeParameter(
...             name='input_file',
...             type=str,
...             required=True,
...             description='Path to input CSV file'
...         ),
...         'delimiter': NodeParameter(
...             name='delimiter',
...             type=str,
...             required=False,
...             default=',',
...             description='CSV delimiter character'
...         )
...     }

Returns:: Dictionary mapping parameter names to their definitions
Return type:: dict[str, NodeParameter]

Used by:

_validate_config(): Validates configuration matches parameters
validate_inputs(): Validates runtime inputs
to_dict(): Includes parameters in serialization
Workflow.connect(): Validates compatible connections

run(**kwargs) → dict[str, Any][source]

Execute the node’s logic.

This is the core method that implements the node’s data processing logic. It receives validated inputs and must return a dictionary of outputs.

Design requirements:

Must be stateless - no side effects between runs
All inputs are provided as keyword arguments
Must return a dictionary (JSON-serializable)
Should handle errors gracefully
Can use self.config for configuration values
Should use self.logger for status reporting

The method is called by execute() which handles:

Input validation before calling run()
Output validation after run() completes
Error wrapping and logging
Execution timing and metrics

Example

>>> def run(self, input_file, delimiter=','):
...     df = pd.read_csv(input_file, delimiter=delimiter)
...     return {
...         'dataframe': df.to_dict(),
...         'row_count': len(df),
...         'columns': list(df.columns)
...     }

Parameters:: **kwargs – Validated input parameters matching get_parameters()
Returns:: Dictionary of outputs that will be validated and passed to downstream nodes
Raises:: NodeExecutionError – If execution fails (will be caught and re-raised by execute())
Return type:: dict[str, Any]

Called by:

execute(): Wraps with validation and error handling
LocalRuntime: During workflow execution
TestRunner: During unit testing

__init__(**kwargs)

Initialize the node with configuration parameters.

This method performs the following initialization steps:

Sets the node ID (defaults to class name)
Creates metadata from provided arguments
Sets up logging for the node
Stores configuration in self.config
Validates configuration against parameters

The configuration is validated by calling _validate_config(), which checks that all required parameters are present and of the correct type.

Parameters:: **kwargs – Configuration parameters including: - id: Optional custom node ID - name: Display name for the node - description: Node description - version: Node version - author: Node author - tags: Set of tags for discovery - Any parameters defined in get_parameters()
Raises:: NodeConfigurationError – If configuration is invalid or if metadata validation fails

Downstream effects:

Creates self.metadata for discovery
Sets up self.logger for execution logging
Stores self.config for runtime access
Validates parameters are correctly specified

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:: None

execute(**runtime_inputs) → dict[str, Any]

Execute the node with validation and error handling.

This is the main entry point for node execution that orchestrates the complete execution lifecycle:

Input validation (validate_inputs)
Execution (run)
Output validation (validate_outputs)
Error handling and logging
Performance metrics

Execution flow:

Logs execution start
Validates inputs against parameter schema
Calls run() with validated inputs
Validates outputs are JSON-serializable
Logs execution time
Returns validated outputs

Error handling strategy:

NodeValidationError: Re-raised as-is (input/output issues)
NodeExecutionError: Re-raised as-is (run() failures)
Other exceptions: Wrapped in NodeExecutionError

Performance tracking:

Records execution start/end times
Logs total execution duration
Includes timing in execution logs

Returns:

Dictionary of validated outputs from run()

Raises:

NodeExecutionError – If execution fails in run()
NodeValidationError – If input/output validation fails

Return type:

dict[str, Any]

Called by:

LocalRuntime: During workflow execution
TaskManager: With execution tracking
Unit tests: For node testing

Downstream effects:

Logs provide execution history
Metrics enable performance monitoring
Validation ensures data integrity

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_output_schema() → dict[str, NodeParameter]

Define output parameters for this node.

This optional method allows nodes to specify their output schema for validation. If not overridden, outputs will only be validated for JSON-serializability.

Design purpose: - Enables static analysis of node outputs - Provides runtime validation of output types - Supports automatic documentation of outputs - Facilitates workflow validation and type checking

The output schema serves similar purposes as input parameters:

Type validation during execution
Documentation for downstream consumers
Workflow connection validation
Export manifest generation

Example

>>> def get_output_schema(self):
...     return {
...         'dataframe': NodeParameter(
...             name='dataframe',
...             type=dict,
...             required=True,
...             description='Processed data as dictionary'
...         ),
...         'row_count': NodeParameter(
...             name='row_count',
...             type=int,
...             required=True,
...             description='Number of rows processed'
...         ),
...         'processing_time': NodeParameter(
...             name='processing_time',
...             type=float,
...             required=False,
...             description='Time taken to process in seconds'
...         )
...     }

Returns:: Dictionary mapping output names to their parameter definitions Empty dict by default (no schema validation)
Return type:: dict[str, NodeParameter]

Used by:

validate_outputs(): Validates runtime outputs
Workflow.connect(): Validates connections between nodes
Documentation generators: Create output documentation
Export systems: Include output schemas in manifests

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

Workflow export
Node persistence
API responses
Configuration sharing

The serialized format includes:

id: Unique node identifier
type: Node class name
metadata: Complete node metadata
config: Current configuration
parameters: Parameter definitions with types

Type serialization:

Python types are converted to string names
Complex types may require custom handling
Parameter defaults are included

Returns:

Node identification and type
Complete metadata
Configuration values
Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

WorkflowExporter: For workflow serialization
CLI: For node inspection
API: For node information endpoints
Debugging: For node state inspection

validate_inputs(**kwargs) → dict[str, Any]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

Gets parameter definitions from get_parameters()
Checks each parameter for:
- Presence (if required)
- Type compatibility
- Null handling for optional parameters
Attempts type conversion if needed
Applies default values for missing optional parameters

Key behaviors:

Required parameters must be provided or have defaults
Optional parameters can be None
Type mismatches attempt conversion before failing
Error messages include parameter descriptions

Example flow:: # Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

Type conversions applied
Defaults for missing optional parameters
None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

execute(): Before passing inputs to run()
Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

ContextFormatterNode

class kailash.nodes.transform.formatters.ContextFormatterNode(**kwargs)[source]

Bases: Node

Formats relevant chunks into context for LLM.

get_parameters() → dict[str, NodeParameter][source]

Define the parameters this node accepts.

This abstract method must be implemented by all concrete nodes to specify their input schema. The parameters define:

What inputs the node expects
Type requirements for each input
Whether inputs are required or optional
Default values for optional inputs
Documentation for each parameter

The returned dictionary is used throughout the node lifecycle:

During initialization: _validate_config() checks configuration
During execution: validate_inputs() validates runtime data
During workflow creation: Used for connection validation
During export: Included in workflow manifests

Example

>>> def get_parameters(self):
...     return {
...         'input_file': NodeParameter(
...             name='input_file',
...             type=str,
...             required=True,
...             description='Path to input CSV file'
...         ),
...         'delimiter': NodeParameter(
...             name='delimiter',
...             type=str,
...             required=False,
...             default=',',
...             description='CSV delimiter character'
...         )
...     }

Returns:: Dictionary mapping parameter names to their definitions
Return type:: dict[str, NodeParameter]

Used by:

_validate_config(): Validates configuration matches parameters
validate_inputs(): Validates runtime inputs
to_dict(): Includes parameters in serialization
Workflow.connect(): Validates compatible connections

run(**kwargs) → dict[str, Any][source]

Execute the node’s logic.

This is the core method that implements the node’s data processing logic. It receives validated inputs and must return a dictionary of outputs.

Design requirements:

Must be stateless - no side effects between runs
All inputs are provided as keyword arguments
Must return a dictionary (JSON-serializable)
Should handle errors gracefully
Can use self.config for configuration values
Should use self.logger for status reporting

The method is called by execute() which handles:

Input validation before calling run()
Output validation after run() completes
Error wrapping and logging
Execution timing and metrics

Example

>>> def run(self, input_file, delimiter=','):
...     df = pd.read_csv(input_file, delimiter=delimiter)
...     return {
...         'dataframe': df.to_dict(),
...         'row_count': len(df),
...         'columns': list(df.columns)
...     }

Parameters:: **kwargs – Validated input parameters matching get_parameters()
Returns:: Dictionary of outputs that will be validated and passed to downstream nodes
Raises:: NodeExecutionError – If execution fails (will be caught and re-raised by execute())
Return type:: dict[str, Any]

Called by:

execute(): Wraps with validation and error handling
LocalRuntime: During workflow execution
TestRunner: During unit testing

__init__(**kwargs)

Initialize the node with configuration parameters.

This method performs the following initialization steps:

Sets the node ID (defaults to class name)
Creates metadata from provided arguments
Sets up logging for the node
Stores configuration in self.config
Validates configuration against parameters

The configuration is validated by calling _validate_config(), which checks that all required parameters are present and of the correct type.

Parameters:: **kwargs – Configuration parameters including: - id: Optional custom node ID - name: Display name for the node - description: Node description - version: Node version - author: Node author - tags: Set of tags for discovery - Any parameters defined in get_parameters()
Raises:: NodeConfigurationError – If configuration is invalid or if metadata validation fails

Downstream effects:

Creates self.metadata for discovery
Sets up self.logger for execution logging
Stores self.config for runtime access
Validates parameters are correctly specified

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:: None

execute(**runtime_inputs) → dict[str, Any]

Execute the node with validation and error handling.

This is the main entry point for node execution that orchestrates the complete execution lifecycle:

Input validation (validate_inputs)
Execution (run)
Output validation (validate_outputs)
Error handling and logging
Performance metrics

Execution flow:

Logs execution start
Validates inputs against parameter schema
Calls run() with validated inputs
Validates outputs are JSON-serializable
Logs execution time
Returns validated outputs

Error handling strategy:

NodeValidationError: Re-raised as-is (input/output issues)
NodeExecutionError: Re-raised as-is (run() failures)
Other exceptions: Wrapped in NodeExecutionError

Performance tracking:

Records execution start/end times
Logs total execution duration
Includes timing in execution logs

Returns:

Dictionary of validated outputs from run()

Raises:

NodeExecutionError – If execution fails in run()
NodeValidationError – If input/output validation fails

Return type:

dict[str, Any]

Called by:

LocalRuntime: During workflow execution
TaskManager: With execution tracking
Unit tests: For node testing

Downstream effects:

Logs provide execution history
Metrics enable performance monitoring
Validation ensures data integrity

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_output_schema() → dict[str, NodeParameter]

Define output parameters for this node.

This optional method allows nodes to specify their output schema for validation. If not overridden, outputs will only be validated for JSON-serializability.

Design purpose: - Enables static analysis of node outputs - Provides runtime validation of output types - Supports automatic documentation of outputs - Facilitates workflow validation and type checking

The output schema serves similar purposes as input parameters:

Type validation during execution
Documentation for downstream consumers
Workflow connection validation
Export manifest generation

Example

>>> def get_output_schema(self):
...     return {
...         'dataframe': NodeParameter(
...             name='dataframe',
...             type=dict,
...             required=True,
...             description='Processed data as dictionary'
...         ),
...         'row_count': NodeParameter(
...             name='row_count',
...             type=int,
...             required=True,
...             description='Number of rows processed'
...         ),
...         'processing_time': NodeParameter(
...             name='processing_time',
...             type=float,
...             required=False,
...             description='Time taken to process in seconds'
...         )
...     }

Returns:: Dictionary mapping output names to their parameter definitions Empty dict by default (no schema validation)
Return type:: dict[str, NodeParameter]

Used by:

validate_outputs(): Validates runtime outputs
Workflow.connect(): Validates connections between nodes
Documentation generators: Create output documentation
Export systems: Include output schemas in manifests

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

Workflow export
Node persistence
API responses
Configuration sharing

The serialized format includes:

id: Unique node identifier
type: Node class name
metadata: Complete node metadata
config: Current configuration
parameters: Parameter definitions with types

Type serialization:

Python types are converted to string names
Complex types may require custom handling
Parameter defaults are included

Returns:

Node identification and type
Complete metadata
Configuration values
Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

WorkflowExporter: For workflow serialization
CLI: For node inspection
API: For node information endpoints
Debugging: For node state inspection

validate_inputs(**kwargs) → dict[str, Any]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

Gets parameter definitions from get_parameters()
Checks each parameter for:
- Presence (if required)
- Type compatibility
- Null handling for optional parameters
Attempts type conversion if needed
Applies default values for missing optional parameters

Key behaviors:

Required parameters must be provided or have defaults
Optional parameters can be None
Type mismatches attempt conversion before failing
Error messages include parameter descriptions

Example flow:: # Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

Type conversions applied
Defaults for missing optional parameters
None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

execute(): Before passing inputs to run()
Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

Logic Nodes 

Note

The Validator node for complex data validation rules is planned for a future release.

Logic nodes control workflow execution flow.

SwitchNode 

class kailash.nodes.logic.operations.SwitchNode(**kwargs)[source]

Bases: Node

Routes data to different outputs based on conditions.

The Switch node enables conditional branching in workflows by evaluating a condition on input data and routing it to different outputs based on the result. This is essential for implementing decision trees, error handling flows, and adaptive processing pipelines.

Design Philosophy:

SwitchNode provides declarative conditional routing without requiring custom logic nodes. It supports both simple boolean conditions and complex multi-case routing, making workflows more maintainable and easier to visualize.

Upstream Dependencies:

Any node producing data that needs conditional routing
Common patterns: validators, analyzers, quality checkers
In cycles: ConvergenceCheckerNode for convergence-based routing

Downstream Consumers:

Different processing nodes based on condition results
MergeNode to rejoin branches after conditional processing
In cycles: nodes that continue or exit based on conditions

Configuration:

condition_field (str): Field in input data to evaluate (for dict inputs) operator (str): Comparison operator (==, !=, >, <, >=, <=, in, contains) value (Any): Value to compare against for boolean conditions cases (list): List of values for multi-case switching case_prefix (str): Prefix for case output fields (default: "case_") pass_condition_result (bool): Include condition result in output

Implementation Details:

Supports both single dict and list of dicts as input
For lists, groups items by condition field value
Multi-case mode creates dynamic outputs (case_X)
Boolean mode uses true_output/false_output
Handles missing fields gracefully

Error Handling:

Missing input_data raises ValueError
Invalid operators return False
Missing condition fields use input directly
Comparison errors caught and return False

Side Effects:

Logs routing decisions for debugging
No external state modifications

Examples

>>> # Simple boolean condition
>>> switch = SwitchNode(condition_field="status", operator="==", value="success")
>>> result = switch.execute(input_data={"status": "success", "data": [1,2,3]})
>>> result["true_output"]
{'status': 'success', 'data': [1, 2, 3]}
>>> result["false_output"] is None
True

>>> # Multi-case switching
>>> switch = SwitchNode(
...     condition_field="priority",
...     cases=["high", "medium", "low"]
... )
>>> result = switch.execute(input_data={"priority": "high", "task": "urgent"})
>>> result["case_high"]
{'priority': 'high', 'task': 'urgent'}

>>> # In cyclic workflows for convergence routing
>>> workflow.add_node("convergence", ConvergenceCheckerNode())
>>> workflow.add_node("switch", SwitchNode(
...     condition_field="converged",
...     operator="==",
...     value=True
... ))
>>> workflow.add_connection("convergence", "result", "switch", "input_data")
>>> # Use CycleBuilder for cyclic connections
>>> cycle = workflow.create_cycle("convergence_loop")
>>> cycle.connect("switch", "false_output", "processor", "input")
>>> cycle.connect("processor", "result", "convergence", "data")
>>> cycle.max_iterations(50).build()
>>> # Non-cyclic output connection
>>> workflow.add_connection("switch", "true_output", "output", "data")

get_parameters() → dict[str, NodeParameter][source]

Define the parameters this node accepts.

This abstract method must be implemented by all concrete nodes to specify their input schema. The parameters define:

What inputs the node expects
Type requirements for each input
Whether inputs are required or optional
Default values for optional inputs
Documentation for each parameter

The returned dictionary is used throughout the node lifecycle:

During initialization: _validate_config() checks configuration
During execution: validate_inputs() validates runtime data
During workflow creation: Used for connection validation
During export: Included in workflow manifests

Example

>>> def get_parameters(self):
...     return {
...         'input_file': NodeParameter(
...             name='input_file',
...             type=str,
...             required=True,
...             description='Path to input CSV file'
...         ),
...         'delimiter': NodeParameter(
...             name='delimiter',
...             type=str,
...             required=False,
...             default=',',
...             description='CSV delimiter character'
...         )
...     }

Returns:: Dictionary mapping parameter names to their definitions
Return type:: dict[str, NodeParameter]

Used by:

_validate_config(): Validates configuration matches parameters
validate_inputs(): Validates runtime inputs
to_dict(): Includes parameters in serialization
Workflow.connect(): Validates compatible connections

get_output_schema() → dict[str, NodeParameter][source]

Define the output schema for SwitchNode.

Note that this returns the standard outputs only. In multi-case mode, additional dynamic outputs (case_X) are created at runtime based on the cases parameter.

Returns:: Standard output parameters
Return type:: Dict[str, NodeParameter]

run(**kwargs) → dict[str, Any][source]

Execute the switch routing logic.

Evaluates conditions on input data and routes to appropriate outputs. Supports both boolean (true/false) and multi-case routing patterns.

Parameters:

**kwargs – Runtime parameters including: input_data (Any): Data to route (required) condition_field (str): Field to check in dict inputs operator (str): Comparison operator value (Any): Value for boolean comparison cases (list): Values for multi-case routing Additional configuration parameters

Returns:

Routing results with keys:

For boolean mode:: true_output: Input data if condition is True false_output: Input data if condition is False condition_result: Boolean result (if enabled)
For multi-case mode:: case_X: Input data for matching cases default: Input data (always present) condition_result: Matched case(s) (if enabled)

Return type:

Dict[str, Any]

Raises:

ValueError – If input_data is not provided

Side Effects:: Logs routing decisions via logger

Examples

>>> switch = SwitchNode()
>>> result = switch.execute(
...     input_data={"score": 85},
...     condition_field="score",
...     operator=">=",
...     value=80
... )
>>> result["true_output"]["score"]
85

__init__(**kwargs)

Initialize the node with configuration parameters.

This method performs the following initialization steps:

Sets the node ID (defaults to class name)
Creates metadata from provided arguments
Sets up logging for the node
Stores configuration in self.config
Validates configuration against parameters

The configuration is validated by calling _validate_config(), which checks that all required parameters are present and of the correct type.

Parameters:: **kwargs – Configuration parameters including: - id: Optional custom node ID - name: Display name for the node - description: Node description - version: Node version - author: Node author - tags: Set of tags for discovery - Any parameters defined in get_parameters()
Raises:: NodeConfigurationError – If configuration is invalid or if metadata validation fails

Downstream effects:

Creates self.metadata for discovery
Sets up self.logger for execution logging
Stores self.config for runtime access
Validates parameters are correctly specified

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:: None

execute(**runtime_inputs) → dict[str, Any]

Execute the node with validation and error handling.

This is the main entry point for node execution that orchestrates the complete execution lifecycle:

Input validation (validate_inputs)
Execution (run)
Output validation (validate_outputs)
Error handling and logging
Performance metrics

Execution flow:

Logs execution start
Validates inputs against parameter schema
Calls run() with validated inputs
Validates outputs are JSON-serializable
Logs execution time
Returns validated outputs

Error handling strategy:

NodeValidationError: Re-raised as-is (input/output issues)
NodeExecutionError: Re-raised as-is (run() failures)
Other exceptions: Wrapped in NodeExecutionError

Performance tracking:

Records execution start/end times
Logs total execution duration
Includes timing in execution logs

Returns:

Dictionary of validated outputs from run()

Raises:

NodeExecutionError – If execution fails in run()
NodeValidationError – If input/output validation fails

Return type:

dict[str, Any]

Called by:

LocalRuntime: During workflow execution
TaskManager: With execution tracking
Unit tests: For node testing

Downstream effects:

Logs provide execution history
Metrics enable performance monitoring
Validation ensures data integrity

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

Workflow export
Node persistence
API responses
Configuration sharing

The serialized format includes:

id: Unique node identifier
type: Node class name
metadata: Complete node metadata
config: Current configuration
parameters: Parameter definitions with types

Type serialization:

Python types are converted to string names
Complex types may require custom handling
Parameter defaults are included

Returns:

Node identification and type
Complete metadata
Configuration values
Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

WorkflowExporter: For workflow serialization
CLI: For node inspection
API: For node information endpoints
Debugging: For node state inspection

validate_inputs(**kwargs) → dict[str, Any]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

Gets parameter definitions from get_parameters()
Checks each parameter for:
- Presence (if required)
- Type compatibility
- Null handling for optional parameters
Attempts type conversion if needed
Applies default values for missing optional parameters

Key behaviors:

Required parameters must be provided or have defaults
Optional parameters can be None
Type mismatches attempt conversion before failing
Error messages include parameter descriptions

Example flow:: # Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

Type conversions applied
Defaults for missing optional parameters
None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

execute(): Before passing inputs to run()
Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

Example Usage:

workflow.add_node("SwitchNode", "route_by_value", config={
    "condition": "customer_segment",
    "routes": {
        "premium": "lifetime_value > 10000",
        "standard": "lifetime_value > 1000",
        "basic": "default"
    }
})

MergeNode 

class kailash.nodes.logic.operations.MergeNode(**kwargs)[source]

Bases: Node

Merges multiple data sources.

This node can combine data from multiple input sources in various ways, making it useful for:

Combining results from parallel branches in a workflow
Joining related data sets
Combining outputs after conditional branching with the SwitchNode
Aggregating collections of data

The merge operation is determined by the merge_type parameter, which supports concat (list concatenation), zip (parallel iteration), and merge_dict (dictionary merging with optional key-based joining for lists of dictionaries).

Example usage:

>>> # Simple list concatenation
>>> merge_node = MergeNode(merge_type="concat")
>>> result = merge_node.execute(data1=[1, 2], data2=[3, 4])
>>> result['merged_data']
[1, 2, 3, 4]

>>> # Dictionary merging
>>> merge_node = MergeNode(merge_type="merge_dict")
>>> result = merge_node.execute(
...     data1={"a": 1, "b": 2},
...     data2={"b": 3, "c": 4}
... )
>>> result['merged_data']
{'a': 1, 'b': 3, 'c': 4}

>>> # List of dicts merging by key
>>> merge_node = MergeNode(merge_type="merge_dict", key="id")
>>> result = merge_node.execute(
...     data1=[{"id": 1, "name": "Alice"}],
...     data2=[{"id": 1, "age": 30}]
... )
>>> result['merged_data']
[{'id': 1, 'name': 'Alice', 'age': 30}]

get_parameters() → dict[str, NodeParameter][source]

Define the parameters this node accepts.

This abstract method must be implemented by all concrete nodes to specify their input schema. The parameters define:

What inputs the node expects
Type requirements for each input
Whether inputs are required or optional
Default values for optional inputs
Documentation for each parameter

The returned dictionary is used throughout the node lifecycle:

During initialization: _validate_config() checks configuration
During execution: validate_inputs() validates runtime data
During workflow creation: Used for connection validation
During export: Included in workflow manifests

Example

>>> def get_parameters(self):
...     return {
...         'input_file': NodeParameter(
...             name='input_file',
...             type=str,
...             required=True,
...             description='Path to input CSV file'
...         ),
...         'delimiter': NodeParameter(
...             name='delimiter',
...             type=str,
...             required=False,
...             default=',',
...             description='CSV delimiter character'
...         )
...     }

Returns:: Dictionary mapping parameter names to their definitions
Return type:: dict[str, NodeParameter]

Used by:

_validate_config(): Validates configuration matches parameters
validate_inputs(): Validates runtime inputs
to_dict(): Includes parameters in serialization
Workflow.connect(): Validates compatible connections

execute(**runtime_inputs) → dict[str, Any][source]

Override execute method for the unknown_merge_type test.

Return type:: dict[str, Any]

run(**kwargs) → dict[str, Any][source]

Execute the node’s logic.

This is the core method that implements the node’s data processing logic. It receives validated inputs and must return a dictionary of outputs.

Design requirements:

Must be stateless - no side effects between runs
All inputs are provided as keyword arguments
Must return a dictionary (JSON-serializable)
Should handle errors gracefully
Can use self.config for configuration values
Should use self.logger for status reporting

The method is called by execute() which handles:

Input validation before calling run()
Output validation after run() completes
Error wrapping and logging
Execution timing and metrics

Example

>>> def run(self, input_file, delimiter=','):
...     df = pd.read_csv(input_file, delimiter=delimiter)
...     return {
...         'dataframe': df.to_dict(),
...         'row_count': len(df),
...         'columns': list(df.columns)
...     }

Parameters:: **kwargs – Validated input parameters matching get_parameters()
Returns:: Dictionary of outputs that will be validated and passed to downstream nodes
Raises:: NodeExecutionError – If execution fails (will be caught and re-raised by execute())
Return type:: dict[str, Any]

Called by:

execute(): Wraps with validation and error handling
LocalRuntime: During workflow execution
TestRunner: During unit testing

__init__(**kwargs)

Initialize the node with configuration parameters.

This method performs the following initialization steps:

Sets the node ID (defaults to class name)
Creates metadata from provided arguments
Sets up logging for the node
Stores configuration in self.config
Validates configuration against parameters

The configuration is validated by calling _validate_config(), which checks that all required parameters are present and of the correct type.

Parameters:: **kwargs – Configuration parameters including: - id: Optional custom node ID - name: Display name for the node - description: Node description - version: Node version - author: Node author - tags: Set of tags for discovery - Any parameters defined in get_parameters()
Raises:: NodeConfigurationError – If configuration is invalid or if metadata validation fails

Downstream effects:

Creates self.metadata for discovery
Sets up self.logger for execution logging
Stores self.config for runtime access
Validates parameters are correctly specified

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:: None

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_output_schema() → dict[str, NodeParameter]

Define output parameters for this node.

This optional method allows nodes to specify their output schema for validation. If not overridden, outputs will only be validated for JSON-serializability.

Design purpose: - Enables static analysis of node outputs - Provides runtime validation of output types - Supports automatic documentation of outputs - Facilitates workflow validation and type checking

The output schema serves similar purposes as input parameters:

Type validation during execution
Documentation for downstream consumers
Workflow connection validation
Export manifest generation

Example

>>> def get_output_schema(self):
...     return {
...         'dataframe': NodeParameter(
...             name='dataframe',
...             type=dict,
...             required=True,
...             description='Processed data as dictionary'
...         ),
...         'row_count': NodeParameter(
...             name='row_count',
...             type=int,
...             required=True,
...             description='Number of rows processed'
...         ),
...         'processing_time': NodeParameter(
...             name='processing_time',
...             type=float,
...             required=False,
...             description='Time taken to process in seconds'
...         )
...     }

Returns:: Dictionary mapping output names to their parameter definitions Empty dict by default (no schema validation)
Return type:: dict[str, NodeParameter]

Used by:

validate_outputs(): Validates runtime outputs
Workflow.connect(): Validates connections between nodes
Documentation generators: Create output documentation
Export systems: Include output schemas in manifests

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

Workflow export
Node persistence
API responses
Configuration sharing

The serialized format includes:

id: Unique node identifier
type: Node class name
metadata: Complete node metadata
config: Current configuration
parameters: Parameter definitions with types

Type serialization:

Python types are converted to string names
Complex types may require custom handling
Parameter defaults are included

Returns:

Node identification and type
Complete metadata
Configuration values
Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

WorkflowExporter: For workflow serialization
CLI: For node inspection
API: For node information endpoints
Debugging: For node state inspection

validate_inputs(**kwargs) → dict[str, Any]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

Gets parameter definitions from get_parameters()
Checks each parameter for:
- Presence (if required)
- Type compatibility
- Null handling for optional parameters
Attempts type conversion if needed
Applies default values for missing optional parameters

Key behaviors:

Required parameters must be provided or have defaults
Optional parameters can be None
Type mismatches attempt conversion before failing
Error messages include parameter descriptions

Example flow:: # Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

Type conversions applied
Defaults for missing optional parameters
None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

execute(): Before passing inputs to run()
Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

Example Usage:

workflow.add_node("MergeNode", "combine_streams", config={
    "strategy": "concat",  # or "join", "union"
    "join_on": "customer_id",  # for join strategy
    "how": "left"  # for join strategy
})

WorkflowNode 

class kailash.nodes.logic.workflow.WorkflowNode(workflow: Workflow | None = None, **kwargs)[source]

Bases: Node

A node that encapsulates and executes an entire workflow.

This node allows workflows to be composed hierarchically, where a complex workflow can be used as a single node within another workflow. This enables powerful composition patterns and reusability.

Design Philosophy: - Workflows become reusable components - Complex logic hidden behind simple interface - Hierarchical composition of workflows - Consistent with standard node behavior

Upstream Components: - Parent workflows that use this node - Workflow builders creating composite workflows - CLI/API creating nested workflow structures

Downstream Usage: - The wrapped workflow and all its nodes - Runtime executing the inner workflow - Results passed to subsequent nodes

Example usage:

>>> # Direct workflow wrapping
>>> from kailash.workflow.graph import Workflow
>>> from kailash.nodes.data.readers import CSVReaderNode
>>> inner_workflow = Workflow("wf-001", "data_processing")
>>> inner_workflow.add_node("reader", CSVReaderNode(file_path="data.csv"))
>>> node = WorkflowNode(workflow=inner_workflow)
>>> node.metadata.name
'WorkflowNode'

>>> # Get parameters from wrapped workflow
>>> params = node.get_parameters()
>>> 'reader_file_path' in params
True
>>> 'inputs' in params
True

>>> # Loading from dictionary
>>> workflow_dict = {
...     "name": "simple",
...     "nodes": {"node1": {"type": "CSVReaderNode", "config": {"file_path": "test.csv"}}},
...     "connections": []
... }
>>> node = WorkflowNode(workflow_dict=workflow_dict)
>>> node._workflow.name
'simple'

Implementation Details: - Parameters derived from workflow entry nodes - Outputs mapped from workflow exit nodes - Uses LocalRuntime for execution - Validates workflow structure on load

Error Handling: - Configuration errors for invalid workflows - Execution errors wrapped with context - Clear error messages for debugging

Side Effects: - Executes entire workflow when run - May create temporary files/state - Logs execution progress

Parameters:: workflow (Workflow | None)

__init__(workflow: Workflow | None = None, **kwargs)

Initialize the WorkflowNode.

Parameters:

workflow (Workflow | None) – Optional workflow instance to wrap
**kwargs – Additional configuration including: - workflow_path: Path to load workflow from file - workflow_dict: Dictionary representation of workflow - name: Display name for the node - description: Node description - input_mapping: Map node inputs to workflow inputs - output_mapping: Map workflow outputs to node outputs

Raises:

NodeConfigurationError – If no workflow source provided or if workflow loading fails

property workflow: Workflow | None: The inner wrapped Workflow (None until loaded).

get_parameters() → dict[str, NodeParameter][source]

Define parameters based on workflow entry nodes.

Analyzes the wrapped workflow to determine required inputs: 1. Finds entry nodes (no incoming connections) 2. Aggregates their parameters 3. Adds generic ‘inputs’ parameter for overrides

Returns:: Dictionary of parameters derived from workflow structure
Return type:: dict[str, NodeParameter]

get_output_schema() → dict[str, NodeParameter][source]

Define output schema based on workflow exit nodes.

Analyzes the wrapped workflow to determine outputs: 1. Finds exit nodes (no outgoing connections) 2. Aggregates their output schemas 3. Includes general ‘results’ output

Returns:: Dictionary of output parameters from workflow structure
Return type:: dict[str, NodeParameter]

run(**kwargs) → dict[str, Any][source]

Execute the wrapped workflow.

Executes the inner workflow with proper input mapping: 1. Maps node inputs to workflow node inputs 2. Executes workflow using LocalRuntime 3. Maps workflow outputs to node outputs

Parameters:

**kwargs – Input parameters for the workflow

Returns:

results: Complete workflow execution results
Mapped outputs from exit nodes

Return type:

Dictionary containing

Raises:

NodeExecutionError – If workflow execution fails

to_dict() → dict[str, Any][source]

Convert node to dictionary representation.

Serializes the WorkflowNode including its wrapped workflow for persistence and export.

Returns:: Dictionary containing node configuration and workflow
Return type:: dict[str, Any]

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:: None

execute(**runtime_inputs) → dict[str, Any]

Execute the node with validation and error handling.

This is the main entry point for node execution that orchestrates the complete execution lifecycle:

Input validation (validate_inputs)
Execution (run)
Output validation (validate_outputs)
Error handling and logging
Performance metrics

Execution flow:

Logs execution start
Validates inputs against parameter schema
Calls run() with validated inputs
Validates outputs are JSON-serializable
Logs execution time
Returns validated outputs

Error handling strategy:

NodeValidationError: Re-raised as-is (input/output issues)
NodeExecutionError: Re-raised as-is (run() failures)
Other exceptions: Wrapped in NodeExecutionError

Performance tracking:

Records execution start/end times
Logs total execution duration
Includes timing in execution logs

Returns:

Dictionary of validated outputs from run()

Raises:

NodeExecutionError – If execution fails in run()
NodeValidationError – If input/output validation fails

Return type:

dict[str, Any]

Called by:

LocalRuntime: During workflow execution
TaskManager: With execution tracking
Unit tests: For node testing

Downstream effects:

Logs provide execution history
Metrics enable performance monitoring
Validation ensures data integrity

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

validate_inputs(**kwargs) → dict[str, Any]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

Gets parameter definitions from get_parameters()
Checks each parameter for:
- Presence (if required)
- Type compatibility
- Null handling for optional parameters
Attempts type conversion if needed
Applies default values for missing optional parameters

Key behaviors:

Required parameters must be provided or have defaults
Optional parameters can be None
Type mismatches attempt conversion before failing
Error messages include parameter descriptions

Example flow:: # Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

Type conversions applied
Defaults for missing optional parameters
None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

execute(): Before passing inputs to run()
Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

Example Usage:

from kailash.workflow import Workflow
from kailash.nodes.logic import WorkflowNode

# Create a reusable workflow
data_processor = Workflow("data_processor")
# ... add nodes to workflow ...

# Wrap it as a node
processor_node = WorkflowNode(workflow=data_processor)

# Use in another workflow
main_workflow = Workflow("main")
main_workflow.add_node("process", processor_node)

# Load from file
file_processor = WorkflowNode(
    workflow_path="workflows/processor.yaml",
    name="file_processor"
)

# Custom parameter mapping
custom_processor = WorkflowNode(
    workflow=data_processor,
    input_mapping={
        "rows": {"node": "reader", "parameter": "num_rows", "type": int}
    },
    output_mapping={
        "count": {"node": "writer", "output": "row_count", "type": int}
    }
)

Validator 

Note

🚧 Coming Soon - This node is planned for a future release.

Planned Features: - Schema validation using JSON Schema - Data quality checks - Custom validation rules - Error reporting and data cleansing

Alternative: Use the PythonCodeNode for custom validation logic in the meantime.

AI/ML Nodes 

AI and machine-learning nodes — LLM agents, embedding generation, text classification, agent-to-agent communication, intelligent orchestration, and the self-organizing agent pool — are provided by the Kaizen framework, a separate Terrene Foundation package built on the Core SDK.

Install Kaizen to use these nodes:

pip install kailash-kaizen

Once installed, the nodes register automatically with the Core SDK runtime and can be added to any workflow by name (for example workflow.add_node("LLMAgentNode", "agent", {...})) or imported directly:

from kaizen.nodes.ai import LLMAgentNode, EmbeddingGeneratorNode

The full node catalogue — LLMAgentNode, EmbeddingGeneratorNode, TextClassifier, the A2A* agent-communication nodes, the intelligent orchestration nodes, and the self-organizing agent pool — and its complete API reference live in the Kaizen documentation.

API Nodes 

Nodes for external API integrations.

HTTPRequestNode 

Example Usage:

from kailash.nodes.api import HTTPRequestNode

# Simple GET request
workflow.add_node("HTTPRequestNode", "fetch_data", config={
    "url": "https://api.example.com/data",
    "method": "GET",
    "headers": {
        "Authorization": "Bearer ${API_TOKEN}"
    },
    "timeout": 30
})

# POST with authentication
workflow.add_node("HTTPRequestNode", "create_resource", config={
    "url": "https://api.example.com/resources",
    "method": "POST",
    "auth_type": "bearer",
    "auth_token": "${API_TOKEN}",
    "json_data": {
        "name": "New Resource",
        "type": "example"
    }
})

RESTClientNode 

Example Usage:

from kailash.nodes.api import RESTClientNode

# GET a resource
workflow.add_node("RESTClientNode", "get_user", config={
    "base_url": "https://api.example.com",
    "resource": "users/{id}",
    "path_params": {"id": "123"},
    "method": "GET",
    "auth_type": "bearer",
    "auth_token": "${API_TOKEN}"
})

# Create a new resource
workflow.add_node("RESTClientNode", "create_user", config={
    "base_url": "https://api.example.com",
    "resource": "users",
    "method": "POST",
    "data": {
        "name": "John Doe",
        "email": "john@example.com"
    },
    "version": "v2"
})

GraphQLClient 

Note

🚧 Coming Soon - This node is planned for a future release.

Planned Features: - GraphQL query and mutation support - Variable binding and validation - Schema introspection - Subscription support

Alternative: Use the PythonCodeNode with GraphQL libraries in the meantime.

MCP Nodes 

Model Context Protocol (MCP) nodes for AI context management.

MCPClient 

Example Usage:

from kailash.nodes.mcp import MCPClient

# List available resources
client = MCPClient()
result = client.run(
    server_config={
        "name": "filesystem-server",
        "command": "python",
        "args": ["-m", "mcp_filesystem"]
    },
    operation="list_resources"
)

# Read a specific resource
result = client.run(
    server_config=server_config,
    operation="read_resource",
    resource_uri="file:///path/to/document.txt"
)

# Call a tool on the server
result = client.run(
    server_config=server_config,
    operation="call_tool",
    tool_name="create_file",
    tool_arguments={
        "path": "/path/to/new_file.txt",
        "content": "Hello, World!"
    }
)

Alert Nodes 

Nodes for notifications and alerting systems.

DiscordAlertNode 

Example Usage:

from kailash.nodes.alerts import DiscordAlertNode

# Basic alert
alert = DiscordAlertNode()
result = alert.run(
    webhook_url="https://discord.com/api/webhooks/...",
    title="System Alert",
    message="Service is running normally",
    alert_type="info"
)

# Rich alert with formatting
result = alert.run(
    webhook_url="${DISCORD_WEBHOOK}",
    title="🚨 Critical Error",
    message="Database connection failed",
    alert_type="critical",
    username="System Monitor",
    mentions=["@here"],
    fields=[
        {"name": "Service", "value": "Database", "inline": True},
        {"name": "Status", "value": "Down", "inline": True},
        {"name": "Error", "value": "Connection timeout", "inline": False}
    ],
    footer_text="Automated alert from monitoring system"
)

# Business metrics alert
result = alert.run(
    webhook_url="${DISCORD_WEBHOOK}",
    title="📊 Daily KPI Report",
    message="Daily performance metrics",
    alert_type="info",
    fields=[
        {"name": "Revenue", "value": "$45,231", "inline": True},
        {"name": "Orders", "value": "127", "inline": True},
        {"name": "Conversion", "value": "3.4%", "inline": True}
    ]
)

Key Features:

Rich Discord Embeds: Automatic color coding, custom fields, mentions
Rate Limiting: Built-in 30 requests/minute sliding window protection
Retry Logic: Exponential backoff with configurable attempts
Environment Variables: Secure webhook URL substitution
Production Ready: Comprehensive error handling and logging

Alert Types and Colors:

info - Blue (0x3498db)
success - Green (0x2ecc71)
warning - Orange (0xf39c12)
error - Red (0xe74c3c)
critical - Dark red (0xc0392b)

Runtime Parameters:

webhook_url (str, required): Discord webhook URL (supports ${DISCORD_WEBHOOK})
title (str, required): Alert title
message (str, optional): Alert message/description
alert_type (str, optional): Alert severity - ‘info’, ‘success’, ‘warning’, ‘error’, ‘critical’
username (str, optional): Custom username for the bot
mentions (list, optional): List of mentions (@everyone, @here, <@user_id>)
fields (list, optional): List of embed fields with name, value, inline properties
footer_text (str, optional): Footer text for the embed
retry_attempts (int, optional): Number of retry attempts (default: 3)
retry_delay (float, optional): Base retry delay in seconds (default: 1.0)

Error Handling:

Automatic retry on rate limits and temporary failures
Graceful degradation on webhook validation errors
Comprehensive logging for debugging and monitoring
Non-blocking execution - alerts won’t crash workflows

Code Nodes 

Nodes for executing custom code.

PythonCodeNode (Enhanced in v0.2.1)

class kailash.nodes.code.python.PythonCodeNode(name: str, code: str | None = None, function: Callable | None = None, class_type: type | None = None, process_method: str | None = None, input_types: dict[str, type] | None = None, output_type: type | None = None, input_schema: dict[str, NodeParameter] | None = None, output_schema: dict[str, NodeParameter] | None = None, description: str | None = None, max_code_lines: int = 10, validate_security: bool = False, sandbox_mode: str = 'restricted', **kwargs)[source]

Bases: Node

Node for executing arbitrary Python code.

This node allows users to execute custom Python code within a workflow. It supports multiple input methods: 1. Direct code string execution 2. Function wrapping 3. Class wrapping 4. File-based code loading

Design Purpose: - Provide maximum flexibility for custom logic - Bridge gap between predefined nodes and custom requirements - Enable rapid prototyping without node development - Support both stateless and stateful processing

Key Features: - Type inference from function signatures - Safe code execution with error handling - Support for external libraries - State management for class-based nodes - AST-based security validation

IMPORTANT - Variable Access Pattern: When using PythonCodeNode with code strings, input parameters are directly available as variables in the execution namespace. Do NOT try to access them through an ‘inputs’ dictionary or use locals()/dir() to check for them.

Correct pattern:: # If ‘query’ is passed as an input parameter, it’s directly available result = {‘processed’: query.upper()} # Direct access to ‘query’
Incorrect patterns:: # These will NOT work: query = inputs.get(‘query’, ‘’) # ‘inputs’ dict doesn’t exist query = locals().get(‘query’, ‘’) # locals() is restricted if ‘query’ in dir(): # dir() is restricted

The node supports two output patterns: 1. Single output: Set a ‘result’ variable with your output data 2. Multiple outputs: Define multiple variables - all become available as outputs

Examples

# Single output (traditional pattern) result = {“processed_data”: data}

# Multiple outputs (NEW - more flexible!) filter_data = {“id”: “user-123”} fields_data = {“name”: “Updated”} status = “success”

Example

>>> # Function-based node
>>> def custom_filter(data: pd.DataFrame, threshold: float) -> pd.DataFrame:
...     return data[data['value'] > threshold]

>>> node = PythonCodeNode.from_function(
...     func=custom_filter,
...     name="threshold_filter"
... )

>>> # Class-based stateful node
>>> class MovingAverage:
...     def __init__(self, window_size: int = 3):
...         self.window_size = window_size
...         self.values = []
...
...     def process(self, value: float) -> float:
...         self.values.append(value)
...         if len(self.values) > self.window_size:
...             self.values.pop(0)
...         return sum(self.values) / len(self.values)

>>> node = PythonCodeNode.from_class(
...     cls=MovingAverage,
...     name="moving_avg"
... )

>>> # Code string node
>>> code = '''
... result = []
... for item in data:
...     if item > threshold:
...         result.append(item * 2)
... '''

>>> node = PythonCodeNode(
...     name="custom_processor",
...     code=code,
...     input_types={'data': list, 'threshold': float},
...     output_type=list
... )

Parameters:

name (str)
code (str | None)
function (Callable | None)
class_type (type | None)
process_method (str | None)
input_types (dict[str, type] | None)
output_type (type | None)
input_schema (dict[str, NodeParameter] | None)
output_schema (dict[str, NodeParameter] | None)
description (str | None)
max_code_lines (int)
validate_security (bool)
sandbox_mode (str)

__init__(name: str, code: str | None = None, function: Callable | None = None, class_type: type | None = None, process_method: str | None = None, input_types: dict[str, type] | None = None, output_type: type | None = None, input_schema: dict[str, NodeParameter] | None = None, output_schema: dict[str, NodeParameter] | None = None, description: str | None = None, max_code_lines: int = 10, validate_security: bool = False, sandbox_mode: str = 'restricted', **kwargs)

Initialize a Python code node.

Parameters:

name (str) – Node name
code (str | None) – Python code string to execute
function (Callable | None) – Python function to wrap
class_type (type | None) – Python class to instantiate
process_method (str | None) – Method name for class-based execution
input_types (dict[str, type] | None) – Dictionary of input names to types
output_type (type | None) – Expected output type
input_schema (dict[str, NodeParameter] | None) – Explicit input parameter schema for validation
output_schema (dict[str, NodeParameter] | None) – Explicit output parameter schema for validation
description (str | None) – Node description
max_code_lines (int) – Maximum lines before warning (default: 10)
validate_security (bool) – If True, validate code security at creation time (default: False)
sandbox_mode (str) – Sandbox enforcement mode. “restricted” (default) enforces module allowlist and AST safety checks. “trusted” bypasses sandbox restrictions, allowing any import. Use “trusted” only for code you control and trust.
**kwargs – Additional node parameters

get_parameters() → dict[str, NodeParameter][source]

Define the parameters this node accepts.

Returns:: Dictionary mapping parameter names to their definitions
Return type:: dict[str, NodeParameter]

validate_inputs(**kwargs) → dict[str, Any][source]

Validate runtime inputs.

For code-based nodes, we accept any inputs since the code can use whatever variables it needs.

Parameters:: **kwargs – Runtime inputs
Returns:: All inputs as-is for code nodes, validated inputs for function/class nodes
Return type:: dict[str, Any]

get_output_schema() → dict[str, NodeParameter][source]

Define output parameters for this node.

Returns:: Dictionary mapping output names to their parameter definitions
Return type:: dict[str, NodeParameter]

run(**kwargs) → dict[str, Any][source]

Execute the node’s logic.

Parameters:: **kwargs – Validated input data
Returns:: Dictionary of outputs
Return type:: dict[str, Any]

classmethod from_function(func: Callable, name: str | None = None, description: str | None = None, input_schema: dict[str, NodeParameter] | None = None, output_schema: dict[str, NodeParameter] | None = None, **kwargs) → PythonCodeNode[source]

Create a node from a Python function.

Parameters:

func (Callable) – Python function to wrap
name (str | None) – Node name (defaults to function name)
description (str | None) – Node description
input_schema (dict[str, NodeParameter] | None) – Explicit input parameter schema for validation
output_schema (dict[str, NodeParameter] | None) – Explicit output parameter schema for validation
**kwargs – Additional node parameters

Returns:

PythonCodeNode instance

Return type:

PythonCodeNode

classmethod from_class(class_type: type, process_method: str | None = None, name: str | None = None, description: str | None = None, input_schema: dict[str, NodeParameter] | None = None, output_schema: dict[str, NodeParameter] | None = None, **kwargs) → PythonCodeNode[source]

Create a node from a Python class.

Parameters:

class_type (type) – Python class to wrap
process_method (str | None) – Method name for processing (auto-detected if not provided)
name (str | None) – Node name (defaults to class name)
description (str | None) – Node description
input_schema (dict[str, NodeParameter] | None) – Explicit input parameter schema for validation
output_schema (dict[str, NodeParameter] | None) – Explicit output parameter schema for validation
**kwargs – Additional node parameters

Returns:

PythonCodeNode instance

Return type:

PythonCodeNode

Create a node from a Python file.

Parameters:

file_path (str | Path) – Path to Python file
function_name (str | None) – Function to use from file
class_name (str | None) – Class to use from file
name (str | None) – Node name
description (str | None) – Node description
input_schema (dict[str, NodeParameter] | None)
output_schema (dict[str, NodeParameter] | None)

Returns:

PythonCodeNode instance

Raises:

NodeConfigurationError – If file cannot be loaded

Return type:

PythonCodeNode

execute_code(inputs: dict[str, Any]) → Any[source]

Execute the code with given inputs.

This is a convenience method that directly executes the code without going through the base node validation.

Parameters:: inputs (dict[str, Any]) – Dictionary of input values
Returns:: Result of code execution
Return type:: Any

get_config() → dict[str, Any][source]

Get node configuration for serialization.

Returns:: Configuration dictionary
Return type:: dict[str, Any]

static list_allowed_modules() → list[str][source]

List all allowed modules for import in PythonCodeNode.

Returns:: Sorted list of allowed module names
Return type:: list[str]

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

is not in _INIT_CAPTURE_EXCLUDE,
is not already present in self.config (subclass may have set it directly via **kwargs forwarding),
has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

static check_module_availability(module_name: str) → dict[str, Any][source]

Check if a module is allowed and available for import.

Parameters:: module_name (str) – Name of the module to check
Returns:: Dictionary with status information
Return type:: dict[str, Any]

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:: None

execute(**runtime_inputs) → dict[str, Any]

Execute the node with validation and error handling.

This is the main entry point for node execution that orchestrates the complete execution lifecycle:

Input validation (validate_inputs)
Execution (run)
Output validation (validate_outputs)
Error handling and logging
Performance metrics

Execution flow:

Logs execution start
Validates inputs against parameter schema
Calls run() with validated inputs
Validates outputs are JSON-serializable
Logs execution time
Returns validated outputs

Error handling strategy:

NodeValidationError: Re-raised as-is (input/output issues)
NodeExecutionError: Re-raised as-is (run() failures)
Other exceptions: Wrapped in NodeExecutionError

Performance tracking:

Records execution start/end times
Logs total execution duration
Includes timing in execution logs

Returns:

Dictionary of validated outputs from run()

Raises:

NodeExecutionError – If execution fails in run()
NodeValidationError – If input/output validation fails

Return type:

dict[str, Any]

Called by:

LocalRuntime: During workflow execution
TaskManager: With execution tracking
Unit tests: For node testing

Downstream effects:

Logs provide execution history
Metrics enable performance monitoring
Validation ensures data integrity

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

enabled: Whether caching is enabled
size: Current cache size
max_size: Maximum cache size
hits: Number of cache hits
misses: Number of cache misses
evictions: Number of cache evictions
hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

key (str) – The key to retrieve from the workflow context
default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:: NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

key (str) – The key to store the value under
value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)

to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

Workflow export
Node persistence
API responses
Configuration sharing

The serialized format includes:

id: Unique node identifier
type: Node class name
metadata: Complete node metadata
config: Current configuration
parameters: Parameter definitions with types

Type serialization:

Python types are converted to string names
Complex types may require custom handling
Parameter defaults are included

Returns:

Node identification and type
Complete metadata
Configuration values
Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

WorkflowExporter: For workflow serialization
CLI: For node inspection
API: For node information endpoints
Debugging: For node state inspection

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

Schema validation: If get_output_schema() is defined, validates types and required fields
JSON serialization: Ensures all outputs can be serialized

Validation process:

Check outputs is a dictionary
If output schema exists:
- Validate required fields are present
- Check type compatibility
- Attempt type conversion if needed
Verify JSON-serializability
Return validated outputs

Schema validation features:

Required outputs must be present
Optional outputs can be None or missing
Type mismatches attempt conversion
Clear error messages with field details

Parameters:: outputs (dict[str, Any]) – Outputs to validate from run() method
Returns:: The same outputs dictionary if valid
Raises:: NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values
Return type:: dict[str, Any]

Called by:

execute(): After run() completes
Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:: patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache
Return type:: None

validate_code(code: str) → dict[str, Any][source]

Validate Python code and provide detailed feedback.

Parameters:: code (str) – Python code to validate
Returns:: Dictionary with validation results
Return type:: dict[str, Any]

Example Usage:

# Inline code execution
workflow.add_node("PythonCodeNode", "process", config={
    "code": '''
import pandas as pd

# Access input data
df = inputs["data"]

# Process data
result = df.groupby("category").agg({
    "revenue": "sum",
    "quantity": "count"
})

# Return results
return {"summary": result}
'''
})

# Execute from file
workflow.add_node("PythonCodeNode", "analyze", config={
    "mode": "file",
    "file_path": "scripts/analysis.py"
})

# Call specific function
workflow.add_node("PythonCodeNode", "transform", config={
    "mode": "function",
    "code": '''
def process_data(df):
    df["processed"] = True
    return df.sort_values("timestamp")
''',
    "function_name": "process_data"
})

Enhanced File Processing (v0.2.1):

PythonCodeNode now supports additional modules for real-world file processing:

# File processing with new modules
file_processor = PythonCodeNode(name="file_processor", code='''
import csv
import pathlib
import mimetypes
import glob
import xml.etree.ElementTree as ET

# Process CSV files
with open(file_path, 'r') as f:
    reader = csv.DictReader(f)
    data = list(reader)

# Detect MIME types
mime_type = mimetypes.guess_type(file_path)[0]

# Modern path operations
path = pathlib.Path(file_path)
file_info = {
    "name": path.name,
    "size": path.stat().st_size,
    "suffix": path.suffix
}

# Pattern matching
related_files = glob.glob(f"{path.parent}/*.{path.suffix[1:]}")

result = {
    "data": data,
    "mime_type": mime_type,
    "file_info": file_info,
    "related_files": related_files
}
''')

New Allowed Modules (v0.2.1):

csv - CSV file processing
mimetypes - MIME type detection
pathlib - Modern path operations
glob - File pattern matching
xml - XML processing

These modules enable real-world data science and file processing workflows while maintaining security restrictions for dangerous operations.

Custom Node Development 

Creating custom nodes is straightforward:

from kailash.nodes import Node, register_node

@register_node("MyCustomNode")
class MyCustomNode(Node):
    """Custom node for specific processing."""

    def validate_config(self) -> None:
        """Validate node configuration."""
        required = ["param1", "param2"]
        for param in required:
            if param not in self.config:
                raise ValueError(f"Missing required parameter: {param}")

    def execute(self, inputs: dict) -> dict:
        """Execute the node logic."""
        # Access configuration
        param1 = self.config["param1"]
        param2 = self.config["param2"]

        # Process inputs
        data = inputs.get("data")
        if data is None:
            raise ValueError("No input data provided")

        # Your custom logic here
        result = self.process_data(data, param1, param2)

        # Return outputs
        return {"processed_data": result}

    def process_data(self, data, param1, param2):
        """Custom processing logic."""
        # Implementation here
        return data

For async operations:

from kailash.nodes import AsyncNode
import aiohttp

@register_node("AsyncAPINode")
class AsyncAPINode(AsyncNode):
    """Async node for API calls."""

    async def execute(self, inputs: dict) -> dict:
        """Execute async API call."""
        url = self.config["url"]

        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                data = await response.json()

        return {"api_response": data}

Node Configuration Best Practices 

Use Environment Variables for Secrets

workflow.add_node("RESTClientNode", "api", config={
    "auth": {
        "token": "${API_TOKEN}"  # Resolved from environment
    }
})

Provide Sensible Defaults

class MyNode(Node):
    def validate_config(self):
        # Set defaults
        self.config.setdefault("timeout", 30)
        self.config.setdefault("retry_count", 3)

Validate Early and Clearly

def validate_config(self):
    if not os.path.exists(self.config["file_path"]):
        raise ValueError(f"File not found: {self.config['file_path']}")

Document Configuration Options

class MyNode(Node):
    """
    My custom node.

    Config:
        file_path (str): Path to input file
        encoding (str, optional): File encoding. Defaults to 'utf-8'
        skip_errors (bool, optional): Skip errors. Defaults to False
    """