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

Database	Type	Nodes/Model	Driver
PostgreSQL	SQL	11	asyncpg
MySQL	SQL	11	aiomysql
SQLite	SQL	11	aiosqlite
MongoDB	Document	8	Motor
pgvector	Vector	3	pgvector

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