| 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
db = DataFlow("sqlite:///app.db", auto_migrate=True)
@db.model
class User:
name: str
email: str
active: bool = True
await db.initialize()
result = await db.express.create("User", {"name": "Alice", "email": "alice@example.com"})
user = await db.express.read("User", result["id"])
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"])
result = db.express_sync.create("User", {"name": "Alice", "email": "alice@example.com"})
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
workflow = WorkflowBuilder()
workflow.add_node("User_Create", "create_user", {
"data": {"name": "John", "email": "john@example.com"}
})
with LocalRuntime() as runtime:
results, run_id = runtime.execute(workflow.build())
user_id = results["create_user"]["result"]
Generated Nodes (11 per model)
Each @db.model class generates:
{Model}_Create - Create single record{Model}_Read - Read by ID{Model}_Update - Update record{Model}_Delete - Delete record{Model}_List - List with filters{Model}_Upsert - Insert or update (atomic){Model}_Count - Efficient COUNT(*) queries{Model}_BulkCreate - Bulk insert{Model}_BulkUpdate - Bulk update{Model}_BulkDelete - Bulk delete{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
Core Operations
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
Advanced Features
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
ML Integration
- dataflow-ml-integration - kailash-ml FeatureStore integration (ConnectionManager, point-in-time queries, polars interop)
Provenance & Audit
Cache Patterns
- cache-cas-fail-closed - CAS (compare-and-swap) fail-closed pattern when primitive can only be satisfied by one backend
Troubleshooting
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
nexus = Nexus(db.get_workflows())
nexus.run()
With Core SDK (Custom Workflows)
from dataflow import DataFlow
from kailash.workflow.builder import WorkflowBuilder
db = DataFlow(connection_string="...")
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
Support
For DataFlow-specific questions, invoke:
dataflow-specialist - DataFlow implementation and patternstesting-specialist - DataFlow testing strategies (NO MOCKING policy)- ``decide-framework
skill - Choose between Core SDK and DataFlow
