القائمة
Kailash DataFlow — MANDATORY for DB/CRUD/bulk/migrations/multi-tenancy. Raw SQL/ORMs BLOCKED.
التثبيت باستخدام Codex أو Claude انسخ هذا Prompt والصقه في Codex أو Claude أو مساعد آخر ليراجع صفحة Skill ويثبّتها لك.
استنادا إلى تصنيف SOC المهني
Claude Code architecture — artifact design, context, agentic patterns. For CC audit/build.
Conformance Walk — freeze-then-judge on the source→delivered→live axis: one cw_core + Source/Delivered/Live adapter families, coverage vs pass-rate, discrete verdicts. Use for any testable surface.
Kailash security (Python) — validation, secrets, injection, authn/z. Hardcoded secrets BLOCKED.
/onboard procedure: read roster + team-memory + posture + claims + codify lease + rules-changed for a new operator joining a multi-operator COC repo.
/certify procedure: brief → probe → gate at 100%; loops failed questions until pass. NO Claude-assistance during gate phase. Curated bank, not LLM-generated.
/ecosystem-init procedure — write the D6 ecosystem-config, run the disclosure scan before write, establish genesis via runEnrollmentCeremony, scaffold non-Kailash STACK.md.
| 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