dataflow — independently scanned and version-tracked by SaferSkills.
SaferSkills independently audited dataflow (Agent Skill) and scored it 100/100 (green). The audit ran 55 deterministic rules across Security, Supply Chain, Maintenance, Transparency, and Community; it found 0 high-severity and 0 lower-severity findings. The full rule-by-rule trace and per-finding evidence are below. Free, methodology-open.
Findings & checks · 0 flagged
Every scanned point with the score it earned and what moved between them.
First recorded scan — no prior version to compare against.
The primary manifest — the file an agent reads to learn what this artifact does.
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": "[email protected]"})
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": "[email protected]"})
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": "[email protected]"}
})
# 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 patternEach @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 platformfrom 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)decide-framework skill` - Choose between Core SDK and DataFlow~30 seconds. Free. No account. Every finding cites a rule and a line of evidence.