Menu
Kailash DataFlow — DB, CRUD, bulk, migrations, multi-tenancy, fabric engine.
Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
| name | dataflow |
| description | Kailash DataFlow — DB, CRUD, bulk, migrations, multi-tenancy, fabric engine. |
DataFlow is a zero-config database framework built on Kailash Core SDK that automatically generates workflow nodes from database models.
Use DataFlow 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.
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": "alice@example.com"})
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": "alice@example.com"})
users = db.express_sync.list("User", {"active": True})
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": "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
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 upsertresults["node_id"]["result"]{} is falsy)if "filter" in kwargs instead of if kwargs.get("filter")db.source()), derived products (@db.product()), fabric runtime (db.start()), 5 source adapters, webhooks, SSRF protection, observability@db.derived_model)db.express.import_file()__validation__ dict)read_url| 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.
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
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", {...})
Use DataFlow when you need to:
For DataFlow-specific questions, invoke:
dataflow-specialist - DataFlow implementation and patternstesting-specialist - DataFlow testing strategies (NO MOCKING policy) skill - Choose between Core SDK and DataFlowBased on SOC occupation classification
Claude Code architecture — artifact design, context, agentic patterns. For CC audit/build.
Kailash Core SDK — workflows, 110+ nodes, runtime, async, cycles, MCP, OpenTelemetry. Use for WorkflowBuilder + connections + runtime patterns.
Kailash DataFlow — MANDATORY for DB/CRUD/bulk/migrations/multi-tenancy. Raw SQL/ORMs BLOCKED.
Kailash Nexus — MANDATORY for HTTP/API/CLI/MCP unified deployment. Direct FastAPI/Flask BLOCKED.
Kailash Kaizen (Python) — MANDATORY for AI agents/RAG/signatures. Custom LLM agents BLOCKED.
Kailash MCP — server/client/tools/resources/auth/transports for AI agent integration.
