Kailash Validation Patterns

Comprehensive validation patterns and compliance checking for Kailash SDK development.

Overview

Validation tools and patterns for:

• Parameter validation
• DataFlow pattern compliance
• Connection validation
• Absolute import checking
• Workflow structure validation
• Security validation

Reference Documentation

Core Validations

Parameter Validation

• validate-parameters - Node parameter validation
  • Required parameters checking
  • Type validation
  • Value range validation
  • Format validation
  • Default value handling

Connection Validation

• validate-connections - Connection validation
  • 4-parameter format check
  • Source/target node existence
  • Parameter name validation
  • Type compatibility
  • Circular dependency detection

Workflow Structure

• validate-workflow-structure - Workflow validation
  • Node ID uniqueness
  • Connection validity
  • Dead-end detection
  • Entry point validation
  • Exit point validation

Framework-Specific Validations

DataFlow Patterns

• validate-dataflow-patterns - DataFlow compliance
  • Result access pattern: results["node_id"]["result"]
  • String ID preservation
  • Multi-instance isolation
  • Transaction patterns
  • Model decorator usage

Absolute Imports

• validate-absolute-imports - Import validation
  • Absolute vs relative imports
  • Module path correctness
  • Circular import detection
  • Missing import detection

Security Validation

• validate-security - Security checks
  • Secret exposure
  • SQL injection risks
  • Code injection risks
  • File path traversal
  • API key handling

Validation Patterns

Parameter Validation Pattern

def validate_node_parameters(node_type: str, params: dict) -> bool:
    """Validate node parameters."""
    required = NODE_REQUIREMENTS[node_type]

    # Check required parameters
    for param in required:
        if param not in params:
            raise ValueError(f"Missing required parameter: {param}")

    # Check types
    for param, value in params.items():
        expected_type = PARAM_TYPES[node_type][param]
        if not isinstance(value, expected_type):
            raise TypeError(f"Invalid type for {param}")

    return True

Connection Validation Pattern

def validate_connection(workflow, source_id, source_param,
                       target_id, target_param) -> bool:
    """Validate 4-parameter connection."""
    # Check nodes exist
    if source_id not in workflow.nodes:
        raise ValueError(f"Source node {source_id} not found")
    if target_id not in workflow.nodes:
        raise ValueError(f"Target node {target_id} not found")

    # Check parameters exist
    source_node = workflow.nodes[source_id]
    if source_param not in source_node.outputs:
        raise ValueError(f"Source param {source_param} not found")

    target_node = workflow.nodes[target_id]
    if target_param not in target_node.inputs:
        raise ValueError(f"Target param {target_param} not found")

    return True

DataFlow Pattern Validation

def validate_dataflow_result_access(code: str) -> bool:
    """Validate DataFlow result access pattern."""
    # ✅ CORRECT: results["node_id"]["result"]
    correct_pattern = r'results\[["\']\w+["\']\]\[["\'](result|data)["\']\]'

    # ❌ WRONG: results["node_id"].result
    wrong_pattern = r'results\[["\']\w+["\']\]\.\w+'

    if re.search(wrong_pattern, code):
        raise ValueError("Use results[id]['result'], not results[id].result")

    if not re.search(correct_pattern, code):
        warnings.warn("Consider using standard result access pattern")

    return True

Security Validation Pattern

def validate_security(workflow) -> list[str]:
    """Run security validations."""
    issues = []

    for node_id, node in workflow.nodes.items():
        # Check for hardcoded secrets
        if has_hardcoded_secrets(node.params):
            issues.append(f"{node_id}: Hardcoded secrets detected")

        # Check for SQL injection risks
        if node.type == "SQLQueryNode":
            if has_sql_injection_risk(node.params.get("query", "")):
                issues.append(f"{node_id}: SQL injection risk")

        # Check for code injection
        if node.type == "PythonCodeNode":
            if has_code_injection_risk(node.params.get("code", "")):
                issues.append(f"{node_id}: Code injection risk")

    return issues

Validation Checklists

Pre-Execution Checklist

• All required parameters provided
• All connections use 4-parameter format
• No missing node IDs
• No duplicate node IDs
• All referenced nodes exist
• No circular dependencies
• Called .build() before execute
• Using correct runtime type

DataFlow Checklist

• Result access uses results["id"]["result"] pattern
• String IDs preserved (no UUID conversion)
• One DataFlow instance per database
• Deferred schema operations enabled
• Transaction boundaries correct
• Model decorators properly applied

Security Checklist

• No hardcoded secrets
• No SQL injection risks
• No code injection risks
• No file path traversal risks
• API keys from environment
• Sensitive data encrypted
• Input validation present

Import Checklist

• All imports are absolute
• No circular imports
• All modules exist
• Import paths correct
• No unused imports

Validation Tools

Automated Validation

from kailash.validation import WorkflowValidator

validator = WorkflowValidator(workflow)

# Run all validations
results = validator.validate_all()

if not results.is_valid:
    for error in results.errors:
        print(f"Error: {error}")
    for warning in results.warnings:
        print(f"Warning: {warning}")

Manual Validation

# Parameter validation
validate_parameters(node_type, params)

# Connection validation
validate_connection(workflow, source_id, source_param,
                   target_id, target_param)

# DataFlow validation
validate_dataflow_patterns(workflow)

# Security validation
issues = validate_security(workflow)

Critical Validation Rules

Must Validate

• ✅ All parameters before execution
• ✅ All connections before building
• ✅ Security risks before deployment
• ✅ Import correctness before commit
• ✅ DataFlow patterns in code review

Never Skip

• ❌ NEVER skip parameter validation
• ❌ NEVER skip connection validation
• ❌ NEVER skip security validation
• ❌ NEVER deploy without validation
• ❌ NEVER commit without import checks

When to Use This Skill

Use this skill when you need to:

• Validate workflow before execution
• Check parameter correctness
• Verify connection format
• Audit security issues
• Review DataFlow patterns
• Check import compliance
• Perform code review
• Ensure gold standards compliance

Integration with Development

Pre-Commit Validation

# Run validation before commit
python -m kailash.validation.cli validate-all

CI/CD Validation

# In CI pipeline
steps:
  - name: Validate Workflows
    run: |
      python -m kailash.validation.cli validate-all
      python -m kailash.validation.cli check-security

Code Review Validation

• Check parameter usage
• Verify connection format
• Audit security issues
• Validate DataFlow patterns
• Check import statements

Related Skills

• 17-gold-standards - Compliance standards
• 15-error-troubleshooting - Error troubleshooting
• 01-core-sdk - Core patterns
• 02-dataflow - DataFlow patterns

Support

For validation help, invoke:

• gold-standards-validator - Compliance checking
• pattern-expert - Pattern validation
• testing-specialist - Test validation
• sdk-navigator - Find validation guides