// Kailash DataFlow — MANDATORY for DB/CRUD/bulk/migrations/multi-tenancy/PostgreSQL/MySQL/MongoDB/pgvector. Raw SQL & ORMs BLOCKED.
[HINT] Download the complete skill directory including SKILL.md and all related files
| name | dataflow |
| description | Kailash DataFlow — MANDATORY for DB/CRUD/bulk/migrations/multi-tenancy. Raw SQL/ORMs BLOCKED. |
DataFlow is a zero-config database framework built on Kailash Core SDK that automatically generates workflow nodes from database models.
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 DataFlow