How jana-2110/data-forge works
Purpose and deployment
DataForge is a no-code ETL and workflow-orchestration system. Users design DAG pipelines visually or generate reviewable pipeline proposals through a local LLM. Docker Compose coordinates the frontend, FastAPI backend, worker, PostgreSQL, and Ollama; optional data targets include MinIO/S3 and DuckDB. Entry points: docker-compose.yml, backend/app/main.py, worker/runner.py, and frontend/src/main.tsx.
Frontend interaction layer
The Vite/React frontend provides the pipeline canvas, run monitoring, connections/triggers, and AI-assisted design. It calls the backend API through a shared client and renders pipeline definitions as editable node/edge graphs. Core surfaces: frontend/src/App.tsx, frontend/src/pages/PipelineDesigner.tsx, frontend/src/pages/AIPipelineBuilder.tsx, frontend/src/lib/api.ts.
Backend API and persistence boundary
FastAPI owns REST endpoints, pipeline lifecycle operations, metadata persistence, and API-to-worker handoff. SQLAlchemy-style models and Alembic migrations define durable pipelines, runs, connections, triggers, templates, and AI conversations/proposals. Core paths: backend/app/main.py, backend/app/api/pipelines.py, backend/app/api/runs.py, backend/app/core/database.py, backend/app/models/pipeline.py, backend/alembic/versions/002_v2_schema.py.
AI proposal pipeline
AI is isolated as a proposal generator, not an execution authority. API requests enter the AI service/agent, which queries read-only metadata tools, invokes a provider selected by the provider factory (default Ollama/Qwen), parses structured pipeline output, validates it, and persists a proposal/diff for explicit approval. Relevant paths: backend/app/api/ai.py, backend/app/ai/services/ai_service.py, backend/app/ai/agents/pipeline_builder.py, backend/app/ai/providers/ollama.py, backend/app/ai/validators/deterministic_validator.py.
Approval and pipeline-definition flow
A proposal becomes executable only after user approval through the AI API. Standard pipeline APIs then create, update, validate, and queue persisted graph definitions. AI schemas separate generated proposal contracts from database models, while the node registry creates a shared allowed-node boundary. Paths: backend/app/ai/schemas/pipeline.py, backend/app/ai/schemas/proposals.py, backend/app/ai/tools/node_registry.py, backend/app/api/pipelines.py.
Execution engine
The separate Python worker consumes queued runs and executes pipeline DAGs. Its DAG builder validates topology/order; the executor dispatches registered node implementations, evaluates expressions, carries data/control-flow state, and records run/node outcomes back through shared persistence. Key paths: worker/runner.py, worker/engine/dag.py, worker/engine/executor.py, worker/engine/expression.py, worker/nodes/registry.py.
Data-processing and orchestration boundary
Worker nodes encapsulate source, transformation, destination, and control-flow behavior. Transformations use Polars; connectors access CSV/files, PostgreSQL, DuckDB, and S3-compatible MinIO. Control-flow nodes support branching, iteration, waits, variables, and child execution, while backend scheduling manages timed triggering. Paths: worker/nodes/transformations.py, worker/nodes/file_db_nodes.py, worker/nodes/pg_source.py, worker/nodes/pg_destination.py, worker/nodes/control_flow.py, backend/app/services/scheduler.py.
Main entry points: database.py PostgreSQL metadata · main.py FastAPI application · docker-compose.yml Docker Compose · ai_service.py AI service
Components 22
Frontend
- React application · Pipeline canvas, monitoring, connections, triggers, and AI design.
frontend/src/App.tsx - Pipeline designer · Edits pipeline node and edge graphs.
frontend/src/pages/PipelineDesigner.tsx - AI pipeline builder
frontend/src/pages/AIPipelineBuilder.tsx - API client
frontend/src/lib/api.ts
Backend API
- FastAPI application · Owns lifecycle operations and worker handoff.
backend/app/main.py - Pipeline API · Creates, updates, validates, and queues graph definitions.
backend/app/api/pipelines.py - Runs API
backend/app/api/runs.py - Scheduler · Manages timed pipeline triggering.
backend/app/services/scheduler.py
AI Proposal System
Generates reviewable proposals only.
- AI API · Requires explicit approval before execution.
backend/app/api/ai.py - AI service
backend/app/ai/services/ai_service.py - Pipeline builder agent · Uses read-only metadata tools to generate structured pipelines.
backend/app/ai/agents/pipeline_builder.py - Ollama provider · Default Qwen-backed provider.
backend/app/ai/providers/ollama.py - Proposal contracts · Separates generated proposals from database models.
backend/app/ai/schemas/proposals.py - Deterministic validator
backend/app/ai/validators/deterministic_validator.py - Allowed node registry · Shared boundary for permitted pipeline node types.
backend/app/ai/tools/node_registry.py
Worker Runtime
- Worker runner · Consumes queued runs.
worker/runner.py - DAG builder · Validates graph topology and execution order.
worker/engine/dag.py - DAG executor · Carries data and control-flow state and records outcomes.
worker/engine/executor.py - Worker node registry · Dispatches source, transform, destination, and control-flow nodes.
worker/nodes/registry.py
Persistence & Targets
- PostgreSQL metadata · Stores pipelines, runs, connections, triggers, templates, and proposals.
backend/app/core/database.py - Data sources & targets · Files, PostgreSQL, DuckDB, and S3-compatible MinIO.
worker/nodes/file_db_nodes.py
Other components
- Docker Compose · Coordinates application services and optional targets.
docker-compose.yml
Connections 28
- Docker Compose → React application: runs
- Docker Compose → FastAPI application: runs
- Docker Compose → Worker runner: runs
- Docker Compose → PostgreSQL metadata: provisions
- React application → Pipeline designer: renders
- React application → AI pipeline builder: renders
- Pipeline designer → API client: uses
- AI pipeline builder → API client: uses
- API client → FastAPI application: REST requests
- FastAPI application → Pipeline API: routes
- FastAPI application → Runs API: routes
- FastAPI application → AI API: routes
- Pipeline API → PostgreSQL metadata: persists definitions and queues runs
- Runs API → PostgreSQL metadata: reads run status
- Scheduler → PostgreSQL metadata: queues triggered runs
- AI API → AI service: submits proposal request
- AI service → Pipeline builder agent: orchestrates
- Pipeline builder agent → Ollama provider: invokes
- Pipeline builder agent → Allowed node registry: checks allowed nodes
- AI service → Proposal contracts: parses structured output
- AI service → Deterministic validator: validates proposal
- Deterministic validator → PostgreSQL metadata: persists proposal and diff
- AI API → Pipeline API: approved proposal becomes definition
- Worker runner → PostgreSQL metadata: claims queued runs and records outcomes
- Worker runner → DAG builder: builds execution plan
- DAG builder → DAG executor: validated order
- DAG executor → Worker node registry: dispatches nodes
- Worker node registry → Data sources & targets: reads and writes data