Dataflow

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

# 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", str(result["id"]))
users = await db.express.list("User", {"active": True})
count = await db.express.count("User")
await db.express.update("User", str(result["id"]), {"name": "Bob"})
await db.express.delete("User", str(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})

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

# 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

Generated Nodes (11 per model)

Each @db.model class generates:

1. {Model}_Create - Create single record
2. {Model}_Read - Read by ID
3. {Model}_Update - Update record
4. {Model}_Delete - Delete record
5. {Model}_List - List with filters
6. {Model}_Upsert - Insert or update (atomic)
7. {Model}_Count - Efficient COUNT(*) queries
8. {Model}_BulkCreate - Bulk insert
9. {Model}_BulkUpdate - Bulk update
10. {Model}_BulkDelete - Bulk delete
11. {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

• dataflow-quickstart - Quick start guide
• dataflow-installation - Installation and setup
• dataflow-models - Defining models with @db.model
• dataflow-connection-config - Database connection

Core Operations

• dataflow-crud-operations - Create, Read, Update, Delete
• dataflow-queries - Query patterns and filtering
• dataflow-aggregation - SQL aggregation queries (COUNT/SUM/AVG/MIN/MAX GROUP BY)
• dataflow-bulk-operations - Batch operations
• dataflow-transactions - Transaction management
• dataflow-connection-isolation - ⚠️ CRITICAL: ACID guarantees

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

• dataflow-derived-models - Application-layer materialized views (@db.derived_model)
• dataflow-file-import - File ingestion (CSV/Excel/Parquet/JSON) + db.express.import_file()
• dataflow-validation-dsl - Declarative validation (__validation__ dict)
• dataflow-express-cache - Model-scoped Express caching with TTL
• dataflow-read-replicas - Read/write splitting with read_url
• dataflow-retention - Data retention (archive/delete/partition policies)
• dataflow-events - Write event emission + Core SDK EventBus integration

Advanced Features

• dataflow-multi-instance - Multiple database instances
• dataflow-multi-tenancy - Multi-tenant architectures
• dataflow-existing-database - Working with existing databases
• dataflow-migrations-quick - Database migrations
• dataflow-custom-nodes - Custom database nodes
• dataflow-sqlite-concurrency - SQLite connection pooling, WAL mode, read/write splitting, memory DB URI patterns

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

• dataflow-connection-config - Pool auto-scaling, env vars, override scenarios
• dataflow-monitoring - Pool utilization, leak detection, health checks, diagnostics

ML Integration

• dataflow-ml-integration - kailash-ml FeatureStore integration (ConnectionManager, point-in-time queries, polars interop)

Troubleshooting

• dataflow-gotchas - Common pitfalls

Database Support Matrix

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

# Auto-generates API + CLI + MCP
nexus = Nexus(db.get_workflows())
nexus.run()  # Instant multi-channel platform

With Core SDK (Custom Workflows)

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", {...})

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

• 01-core-sdk - Core workflow patterns (canonical node pattern)
• 03-nexus - Multi-channel deployment
• 04-kaizen - AI agent integration
• 17-gold-standards - Best practices

Support

For DataFlow-specific questions, invoke:

• dataflow-specialist - DataFlow implementation and patterns
• testing-specialist - DataFlow testing strategies (NO MOCKING policy)
• ``decide-framework skill - Choose between Core SDK and DataFlow

Converted and distributed by TomeVault — claim your Tome and manage your conversions.