Dataflow

Kailash DataFlow - Zero-Config Database Framework

DataFlow is a zero-config database framework built on Kailash Core SDK that automatically generates workflow nodes from database models.

Overview

• Automatic Node Generation: 11 nodes per model (@db.model decorator)
• Multi-Database Support: PostgreSQL, MySQL, SQLite (SQL) + MongoDB (Document) + pgvector (Vector Search)
• Enterprise Features: Multi-tenancy, multi-instance isolation, transactions
• Zero Configuration: String IDs preserved, deferred schema operations
• Developer Experience: Enhanced errors (DF-XXX codes), strict mode validation, debug agent, CLI tools

Quick Start

DataFlow nodes follow the canonical 4-parameter pattern from /01-core-sdk.

from dataflow import DataFlow
from kailash.workflow.builder import WorkflowBuilder
from kailash.runtime.local import LocalRuntime

# Initialize DataFlow
db = DataFlow(connection_string="postgresql://user:pass@localhost/db")

# Define model (generates 11 nodes automatically)
@db.model
class User:
    id: str  # String IDs preserved
    name: str
    email: str

# Use generated nodes in workflows
workflow = WorkflowBuilder()
workflow.add_node("User_Create", "create_user", {
    "data": {"name": "John", "email": "john@example.com"}
})

# Execute with context manager (recommended for resource cleanup)
with LocalRuntime() as runtime:
    results, run_id = runtime.execute(workflow.build())
    user_id = results["create_user"]["result"]  # Access pattern

Generated Nodes (11 per model)

Each @db.model class generates:

1. {Model}_Create - Create single record
2. {Model}_Read - Read by ID
3. {Model}_Update - Update record
4. {Model}_Delete - Delete record
5. {Model}_List - List with filters
6. {Model}_Upsert - Insert or update (atomic)
7. {Model}_Count - Efficient COUNT(*) queries
8. {Model}_BulkCreate - Bulk insert
9. {Model}_BulkUpdate - Bulk update
10. {Model}_BulkDelete - Bulk delete
11. {Model}_BulkUpsert - Bulk upsert

Critical Rules

• ✅ String IDs preserved (no UUID conversion)
• ✅ Deferred schema operations (safe for Docker/FastAPI)
• ✅ Multi-instance isolation (one DataFlow per database)
• ✅ Result access: results["node_id"]["result"]
• ❌ NEVER use truthiness checks on filter/data parameters (empty dict {} is falsy)
• ❌ ALWAYS use key existence checks: if "filter" in kwargs instead of if kwargs.get("filter")
• ❌ NEVER use direct SQL when DataFlow nodes exist
• ❌ NEVER use SQLAlchemy/Django ORM alongside DataFlow

Reference Documentation

Getting Started

• dataflow-quickstart - Quick start guide
• dataflow-installation - Installation and setup
• dataflow-models - Defining models with @db.model
• dataflow-connection-config - Database connection

Core Operations

• dataflow-crud-operations - Create, Read, Update, Delete
• dataflow-queries - Query patterns and filtering
• dataflow-bulk-operations - Batch operations
• dataflow-transactions - Transaction management
• dataflow-connection-isolation - ⚠️ CRITICAL: ACID guarantees

Advanced Features

• dataflow-multi-instance - Multiple database instances
• dataflow-multi-tenancy - Multi-tenant architectures
• dataflow-existing-database - Working with existing databases
• dataflow-migrations-quick - Database migrations
• dataflow-custom-nodes - Custom database nodes

Developer Experience Tools

• dataflow-strict-mode - Build-time validation (4-layer, OFF/WARN/STRICT)
• dataflow-debug-agent - Intelligent error analysis (5-stage pipeline)
• ErrorEnhancer - Automatic error enhancement (40+ DF-XXX codes)
• Inspector API - Self-service debugging (18 introspection methods)
• CLI Tools - dataflow-validate, dataflow-analyze, dataflow-debug (5 commands)

Troubleshooting

• create-vs-update guide - CreateNode vs UpdateNode
• top-10-errors - Quick fix for 90% of issues
• dataflow-gotchas - Common pitfalls

Database Support Matrix

Not an ORM: DataFlow generates workflow nodes, not ORM models. Uses string-based result access and integrates with Kailash's workflow execution model.

Integration Patterns

With Nexus (Multi-Channel)

from dataflow import DataFlow
from nexus import Nexus

db = DataFlow(connection_string="...")
@db.model
class User:
    id: str
    name: str

# Auto-generates API + CLI + MCP
nexus = Nexus(db.get_workflows())
nexus.run()  # Instant multi-channel platform

With Core SDK (Custom Workflows)

from dataflow import DataFlow
from kailash.workflow.builder import WorkflowBuilder

db = DataFlow(connection_string="...")
# Use db-generated nodes in custom workflows
workflow = WorkflowBuilder()
workflow.add_node("User_Create", "user1", {...})

When to Use This Skill

Use DataFlow when you need to:

• Perform database operations in workflows
• Generate CRUD APIs automatically (with Nexus)
• Implement multi-tenant systems
• Work with existing databases
• Build database-first applications
• Handle bulk data operations

Related Skills

• 01-core-sdk - Core workflow patterns (canonical node pattern)
• 03-nexus - Multi-channel deployment
• 04-kaizen - AI agent integration
• 17-gold-standards - Best practices

Support

For DataFlow-specific questions, invoke:

• dataflow-specialist - DataFlow implementation and patterns
• testing-specialist - DataFlow testing strategies (NO MOCKING policy)
• framework-advisor - Choose between Core SDK and DataFlow