# jana-2110/data-forge architecture

> Architecture diagram of the GitHub repository jana-2110/data-forge, generated by GitDiagram from its file tree, README and a sample of source files (last updated 2026-08-14). It is an AI-made overview: check details against the source.

- Interactive diagram: https://gitdiagram.com/jana-2110/data-forge
- Repository: https://github.com/jana-2110/data-forge

## Overview

#### 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: [`backend/app/core/database.py`](https://github.com/jana-2110/data-forge/blob/main/backend/app/core/database.py) (PostgreSQL metadata), [`backend/app/main.py`](https://github.com/jana-2110/data-forge/blob/main/backend/app/main.py) (FastAPI application), [`docker-compose.yml`](https://github.com/jana-2110/data-forge/blob/main/docker-compose.yml) (Docker Compose), [`backend/app/ai/services/ai_service.py`](https://github.com/jana-2110/data-forge/blob/main/backend/app/ai/services/ai_service.py) (AI service)

## Diagram (Mermaid)

```mermaid
flowchart TD

subgraph group_frontend["Frontend"]
  node_ui["React application<br/>Vite/React UI<br/>[App.tsx]"]
  node_designer["Pipeline designer<br/>canvas surface"]
  node_ai_builder_ui["AI pipeline builder<br/>AI design surface"]
  node_api_client["API client<br/>frontend client<br/>[api.ts]"]
end

subgraph group_backend["Backend API"]
  node_api["FastAPI application<br/>REST API<br/>[main.py]"]
  node_pipeline_api["Pipeline API<br/>pipeline endpoints<br/>[pipelines.py]"]
  node_runs_api["Runs API<br/>run endpoints<br/>[runs.py]"]
  node_scheduler["Scheduler<br/>[scheduler.py]"]
end

subgraph group_ai["AI Proposal System"]
  node_ai_api["AI API<br/>proposal endpoints<br/>[ai.py]"]
  node_ai_service["AI service<br/>proposal orchestration<br/>[ai_service.py]"]
  node_pipeline_agent["Pipeline builder agent<br/>LLM agent"]
  node_ollama["Ollama provider<br/>local LLM provider<br/>[ollama.py]"]
  node_proposal_contracts["Proposal contracts<br/>AI schemas<br/>[proposals.py]"]
  node_validator["Deterministic validator<br/>proposal validation"]
  node_node_registry["Allowed node registry<br/>node policy<br/>[node_registry.py]"]
end

subgraph group_worker["Worker Runtime"]
  node_worker_runner["Worker runner<br/>run consumer<br/>[runner.py]"]
  node_dag_builder["DAG builder<br/>topology validation<br/>[dag.py]"]
  node_executor["DAG executor<br/>node dispatcher<br/>[executor.py]"]
  node_worker_registry["Worker node registry<br/>node implementations<br/>[registry.py]"]
end

subgraph group_data["Persistence &amp; Targets"]
  node_database[("PostgreSQL metadata<br/>durable persistence<br/>[database.py]")]
  node_data_targets["Data sources &amp; targets<br/>connectors<br/>[file_db_nodes.py]"]
end

node_compose{{"Docker Compose<br/>deployment<br/>[docker-compose.yml]"}}

node_compose -->|"runs"| node_ui
node_compose -->|"runs"| node_api
node_compose -->|"runs"| node_worker_runner
node_compose -->|"provisions"| node_database
node_ui -->|"renders"| node_designer
node_ui -->|"renders"| node_ai_builder_ui
node_designer -->|"uses"| node_api_client
node_ai_builder_ui -->|"uses"| node_api_client
node_api_client -->|"REST requests"| node_api
node_api -->|"routes"| node_pipeline_api
node_api -->|"routes"| node_runs_api
node_api -->|"routes"| node_ai_api
node_pipeline_api -->|"persists definitions and queues runs"| node_database
node_runs_api -->|"reads run status"| node_database
node_scheduler -->|"queues triggered runs"| node_database
node_ai_api -->|"submits proposal request"| node_ai_service
node_ai_service -->|"orchestrates"| node_pipeline_agent
node_pipeline_agent -->|"invokes"| node_ollama
node_pipeline_agent -->|"checks allowed nodes"| node_node_registry
node_ai_service -->|"parses structured output"| node_proposal_contracts
node_ai_service -->|"validates proposal"| node_validator
node_validator -->|"persists proposal and diff"| node_database
node_ai_api -->|"approved proposal becomes definition"| node_pipeline_api
node_worker_runner -->|"claims queued runs and records outcomes"| node_database
node_worker_runner -->|"builds execution plan"| node_dag_builder
node_dag_builder -->|"validated order"| node_executor
node_executor -->|"dispatches nodes"| node_worker_registry
node_worker_registry -->|"reads and writes data"| node_data_targets

click node_compose "https://github.com/jana-2110/data-forge/blob/main/docker-compose.yml"
click node_ui "https://github.com/jana-2110/data-forge/blob/main/frontend/src/App.tsx"
click node_designer "https://github.com/jana-2110/data-forge/blob/main/frontend/src/pages/PipelineDesigner.tsx"
click node_ai_builder_ui "https://github.com/jana-2110/data-forge/blob/main/frontend/src/pages/AIPipelineBuilder.tsx"
click node_api_client "https://github.com/jana-2110/data-forge/blob/main/frontend/src/lib/api.ts"
click node_api "https://github.com/jana-2110/data-forge/blob/main/backend/app/main.py"
click node_pipeline_api "https://github.com/jana-2110/data-forge/blob/main/backend/app/api/pipelines.py"
click node_runs_api "https://github.com/jana-2110/data-forge/blob/main/backend/app/api/runs.py"
click node_scheduler "https://github.com/jana-2110/data-forge/blob/main/backend/app/services/scheduler.py"
click node_ai_api "https://github.com/jana-2110/data-forge/blob/main/backend/app/api/ai.py"
click node_ai_service "https://github.com/jana-2110/data-forge/blob/main/backend/app/ai/services/ai_service.py"
click node_pipeline_agent "https://github.com/jana-2110/data-forge/blob/main/backend/app/ai/agents/pipeline_builder.py"
click node_ollama "https://github.com/jana-2110/data-forge/blob/main/backend/app/ai/providers/ollama.py"
click node_proposal_contracts "https://github.com/jana-2110/data-forge/blob/main/backend/app/ai/schemas/proposals.py"
click node_validator "https://github.com/jana-2110/data-forge/blob/main/backend/app/ai/validators/deterministic_validator.py"
click node_node_registry "https://github.com/jana-2110/data-forge/blob/main/backend/app/ai/tools/node_registry.py"
click node_database "https://github.com/jana-2110/data-forge/blob/main/backend/app/core/database.py"
click node_worker_runner "https://github.com/jana-2110/data-forge/blob/main/worker/runner.py"
click node_dag_builder "https://github.com/jana-2110/data-forge/blob/main/worker/engine/dag.py"
click node_executor "https://github.com/jana-2110/data-forge/blob/main/worker/engine/executor.py"
click node_worker_registry "https://github.com/jana-2110/data-forge/blob/main/worker/nodes/registry.py"
click node_data_targets "https://github.com/jana-2110/data-forge/blob/main/worker/nodes/file_db_nodes.py"

classDef toneNeutral fill:#f8fafc,stroke:#334155,stroke-width:1.5px,color:#0f172a
classDef toneBlue fill:#dbeafe,stroke:#2563eb,stroke-width:1.5px,color:#172554
classDef toneAmber fill:#fef3c7,stroke:#d97706,stroke-width:1.5px,color:#78350f
classDef toneMint fill:#dcfce7,stroke:#16a34a,stroke-width:1.5px,color:#14532d
classDef toneRose fill:#ffe4e6,stroke:#e11d48,stroke-width:1.5px,color:#881337
classDef toneIndigo fill:#e0e7ff,stroke:#4f46e5,stroke-width:1.5px,color:#312e81
classDef toneTeal fill:#ccfbf1,stroke:#0f766e,stroke-width:1.5px,color:#134e4a
class node_ui,node_designer,node_ai_builder_ui,node_api_client toneBlue
class node_api,node_pipeline_api,node_runs_api,node_scheduler toneAmber
class node_ai_api,node_ai_service,node_pipeline_agent,node_ollama,node_proposal_contracts,node_validator,node_node_registry toneMint
class node_worker_runner,node_dag_builder,node_executor,node_worker_registry toneRose
class node_database,node_data_targets toneIndigo
class node_compose toneNeutral
```

## Components

### Frontend

- [React application](https://github.com/jana-2110/data-forge/blob/main/frontend/src/App.tsx): Vite/React UI · Pipeline canvas, monitoring, connections, triggers, and AI design. · `frontend/src/App.tsx`
- [Pipeline designer](https://github.com/jana-2110/data-forge/blob/main/frontend/src/pages/PipelineDesigner.tsx): canvas surface · Edits pipeline node and edge graphs. · `frontend/src/pages/PipelineDesigner.tsx`
- [AI pipeline builder](https://github.com/jana-2110/data-forge/blob/main/frontend/src/pages/AIPipelineBuilder.tsx): AI design surface · `frontend/src/pages/AIPipelineBuilder.tsx`
- [API client](https://github.com/jana-2110/data-forge/blob/main/frontend/src/lib/api.ts): frontend client · `frontend/src/lib/api.ts`

### Backend API

- [FastAPI application](https://github.com/jana-2110/data-forge/blob/main/backend/app/main.py): REST API · Owns lifecycle operations and worker handoff. · `backend/app/main.py`
- [Pipeline API](https://github.com/jana-2110/data-forge/blob/main/backend/app/api/pipelines.py): pipeline endpoints · Creates, updates, validates, and queues graph definitions. · `backend/app/api/pipelines.py`
- [Runs API](https://github.com/jana-2110/data-forge/blob/main/backend/app/api/runs.py): run endpoints · `backend/app/api/runs.py`
- [Scheduler](https://github.com/jana-2110/data-forge/blob/main/backend/app/services/scheduler.py): trigger scheduler · Manages timed pipeline triggering. · `backend/app/services/scheduler.py`

### AI Proposal System

Generates reviewable proposals only.

- [AI API](https://github.com/jana-2110/data-forge/blob/main/backend/app/api/ai.py): proposal endpoints · Requires explicit approval before execution. · `backend/app/api/ai.py`
- [AI service](https://github.com/jana-2110/data-forge/blob/main/backend/app/ai/services/ai_service.py): proposal orchestration · `backend/app/ai/services/ai_service.py`
- [Pipeline builder agent](https://github.com/jana-2110/data-forge/blob/main/backend/app/ai/agents/pipeline_builder.py): LLM agent · Uses read-only metadata tools to generate structured pipelines. · `backend/app/ai/agents/pipeline_builder.py`
- [Ollama provider](https://github.com/jana-2110/data-forge/blob/main/backend/app/ai/providers/ollama.py): local LLM provider · Default Qwen-backed provider. · `backend/app/ai/providers/ollama.py`
- [Proposal contracts](https://github.com/jana-2110/data-forge/blob/main/backend/app/ai/schemas/proposals.py): AI schemas · Separates generated proposals from database models. · `backend/app/ai/schemas/proposals.py`
- [Deterministic validator](https://github.com/jana-2110/data-forge/blob/main/backend/app/ai/validators/deterministic_validator.py): proposal validation · `backend/app/ai/validators/deterministic_validator.py`
- [Allowed node registry](https://github.com/jana-2110/data-forge/blob/main/backend/app/ai/tools/node_registry.py): node policy · Shared boundary for permitted pipeline node types. · `backend/app/ai/tools/node_registry.py`

### Worker Runtime

- [Worker runner](https://github.com/jana-2110/data-forge/blob/main/worker/runner.py): run consumer · Consumes queued runs. · `worker/runner.py`
- [DAG builder](https://github.com/jana-2110/data-forge/blob/main/worker/engine/dag.py): topology validation · Validates graph topology and execution order. · `worker/engine/dag.py`
- [DAG executor](https://github.com/jana-2110/data-forge/blob/main/worker/engine/executor.py): node dispatcher · Carries data and control-flow state and records outcomes. · `worker/engine/executor.py`
- [Worker node registry](https://github.com/jana-2110/data-forge/blob/main/worker/nodes/registry.py): node implementations · Dispatches source, transform, destination, and control-flow nodes. · `worker/nodes/registry.py`

### Persistence & Targets

- [PostgreSQL metadata](https://github.com/jana-2110/data-forge/blob/main/backend/app/core/database.py): durable persistence · Stores pipelines, runs, connections, triggers, templates, and proposals. · `backend/app/core/database.py`
- [Data sources & targets](https://github.com/jana-2110/data-forge/blob/main/worker/nodes/file_db_nodes.py): connectors · Files, PostgreSQL, DuckDB, and S3-compatible MinIO. · `worker/nodes/file_db_nodes.py`

### Other components

- [Docker Compose](https://github.com/jana-2110/data-forge/blob/main/docker-compose.yml): deployment · Coordinates application services and optional targets. · `docker-compose.yml`

## Connections

- 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