dataflow

name

dataflow

description

Kailash DataFlow - zero-config data operations framework with automatic model-to-node generation and Data Fabric Engine. Use when asking about 'database operations', 'DataFlow', 'database models', 'CRUD operations', 'bulk operations', 'database queries', 'database migrations', 'multi-tenancy', 'multi-instance', 'database transactions', 'PostgreSQL', 'MySQL', 'SQLite', 'MongoDB', 'pgvector', 'vector search', 'document database', 'RAG', 'semantic search', 'existing database', 'database performance', 'database deployment', 'database testing', 'TDD with databases', 'external data sources', 'data products', 'db.source', 'db.product', 'db.start', 'fabric engine', 'source adapters', 'REST source', 'webhooks', or 'data fabric'. DataFlow is NOT an ORM - it generates 11 workflow nodes per SQL model, 8 nodes for MongoDB, and 3 nodes for vector operations.

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

Express API (Recommended for Simple CRUD)

from dataflow import DataFlow

# Zero-config initialization
db = DataFlow("sqlite:///app.db", auto_migrate=True)

@db.model
class User:
    name: str
    email: str
    active: bool = True

await db.initialize()

# Async Express (default) — 23x faster than workflow primitives
result = await db.express.create("User", {"name": "Alice", "email": "[e-mail]"})
user = await db.express.read("User", result["id"])  # accepts both str and int IDs
users = await db.express.list("User", {"active": True})
count = await db.express.count("User")
await db.express.update("User", result["id"], {"name": "Bob"})
await db.express.delete("User", result["id"])

# Sync Express (CLI scripts, non-async contexts)
result = db.express_sync.create("User", {"name": "Alice", "email": "[e-mail]"})
users = db.express_sync.list("User", {"active": True})

Workflow API (For Multi-Step Operations)

Use WorkflowBuilder only when you need multiple nodes with data flow between them.

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

# Multi-node workflow with connections
workflow = WorkflowBuilder()
workflow.add_node("User_Create", "create_user", {
    "data": {"name": "John", "email": "[e-mail]"}
})

# 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/async)
• ✅ 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-aggregation - SQL aggregation queries (COUNT/SUM/AVG/MIN/MAX GROUP BY)
• dataflow-bulk-operations - Batch operations
• dataflow-transactions - Transaction management
• dataflow-connection-isolation - ⚠️ CRITICAL: ACID guarantees

Advanced Features

Data Fabric Engine

• dataflow-fabric-engine - External data sources (db.source()), derived products (@db.product()), fabric runtime (db.start()), 5 source adapters, webhooks, SSRF protection, observability

Enterprise Features

• dataflow-derived-models - Application-layer materialized views (@db.derived_model)
• dataflow-file-import - File ingestion (CSV/Excel/Parquet/JSON) + db.express.import_file()
• dataflow-validation-dsl - Declarative validation (__validation__ dict)
• dataflow-express-cache - Model-scoped Express caching with TTL
• dataflow-read-replicas - Read/write splitting with read_url
• dataflow-retention - Data retention (archive/delete/partition policies)
• dataflow-events - Write event emission + Core SDK EventBus integration

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
• dataflow-sqlite-concurrency - SQLite connection pooling, WAL mode, read/write splitting, memory DB URI patterns

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)

Connection Pool & Monitoring

• dataflow-connection-config - Pool auto-scaling, env vars, override scenarios
• dataflow-monitoring - Pool utilization, leak detection, health checks, diagnostics

ML Integration

• dataflow-ml-integration - kailash-ml FeatureStore integration (ConnectionManager, point-in-time queries, polars interop)

Provenance & Audit

• dataflow-provenance-audit - Provenance[T] field tracking, audit trail persistence, EventStoreBackend
• dataflow-fabric-cache-consumers - Fabric cache control, consumer adapters, MCP tool generation

Cache Patterns

• cache-cas-fail-closed - CAS (compare-and-swap) fail-closed pattern when primitive can only be satisfied by one backend

Troubleshooting

• 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)
• ``decide-framework skill - Choose between Core SDK and DataFlow