FlowForge

Database-driven pipeline orchestrator. Configure everything from the UI — DB procedures, reports, email (Gmail/M365/SMTP), Google Drive, OneDrive, SFTP, smart attachments, scheduling. No YAML. No Airflow complexity.

What is FlowForge?

FlowForge runs ordered data pipelines: call database procedures, run queries, generate reports (Excel/PDF/CSV/JSON), send emails with smart attachment handling, upload files to Google Drive or OneDrive, and transfer files over SFTP. Everything is configured through a web frontend — no config files to edit, no Python DAGs to write.

Designed for solo developers, data analysts, and small teams who need lightweight automation without the overhead of enterprise orchestration tools.

Features

Pipeline Engine

• No-YAML config — pipelines, reports, email templates, and schedules are all managed in the UI
• Multi-project workspace — organize resources by team or department in one instance, with team-scoped project membership
• 17 step types — db_procedure, db_query, report, email, drive_upload, onedrive_upload, s3_upload, azure_blob_upload, data_load, bulk_load, sftp_transfer, ssh_command, ssh_health_check, db_health_check, data_report, notification (Slack/Teams/Telegram), ai_analyze
• Plugin system — drop a custom BaseStep subclass into FLOWFORGE_PLUGIN_DIR to add community/internal step types without forking
• Pipeline dependencies — chain pipelines so a downstream one launches automatically after all its upstreams succeed; cycle detection included
• Parallel step execution — group steps with parallel_group to run them concurrently in a pipeline
• Step-level DAG dependencies — draw real dependency edges between steps on the pipeline canvas for non-linear execution; a failed stop step only halts its own descendants (independent branches keep running), and {{ steps.* }} context is scoped to a step's transitive ancestors. A pipeline with no drawn edges runs through the original sequential/parallel-group engine, untouched
• Environment promotion — clone a pipeline (steps + non-secret variables) into another project
• Run diff view — compare a run against the previous successful run: row count, duration, and file size deltas per step
• Step retry — configurable retry count (0–10) and backoff delay per step
• On-error control — per-step stop or continue on failure
• Failure webhook — POST notification to any URL when a pipeline fails
• Pipeline clone — duplicate a pipeline with one click
• YAML import/export — from UI or CLI; useful for version control and migration

Databases & Storage

• Database support: PostgreSQL, Oracle (python-oracledb, thin mode — no Instant Client), MySQL / MariaDB, MSSQL, generic ODBC, Snowflake, BigQuery, Redshift
• Google Drive: upload files, create shareable links, smart attachment fallback
• OneDrive / SharePoint: upload files via Microsoft Graph API, chunked upload for large files, shareable links
• AWS S3 / Azure Blob Storage: upload step for either, alongside Drive/OneDrive smart-attachment fallback
• SFTP: download or upload files; password or private-key auth (RSA / ECDSA / Ed25519)
• SSH: run remote commands or collect health metrics (load, memory, disk, top processes) over a single SSH session

Email & Reports

• Email providers: Gmail (OAuth2), Microsoft 365 (MSAL + Graph API), SMTP (any server), SendGrid, AWS SES, Mailgun
• Chat notifications: Slack, Microsoft Teams, and Telegram via the notification step
• Smart attachments: files over a configurable size threshold are automatically uploaded to Drive, OneDrive, S3, or Azure Blob; the email body gets a shareable link instead
• Query results in email: embed live SQL query results — HTML table, key-value summary, or custom Jinja2 — directly in email bodies
• Report formats: Excel (openpyxl, optional template, conditional column formatting), CSV, PDF (WeasyPrint), JSON

Multi-User & Access Control

• Roles: admin, editor, viewer — enforced on every API route and in the UI
• User management: admin UI to create users, change roles, delete accounts
• MFA: TOTP-based two-factor auth with QR code enrollment and 10 backup codes
• SSO: Google OAuth2, Microsoft OAuth2, and SAML 2.0 (Okta, Azure AD, PingFederate)
• Self-service password reset: email-based reset flow, plus self-service password change for any authenticated user
• IP allowlisting: restrict /api/* to a configured CIDR list
• JWT with revocation: server-side token blocklist; POST /api/auth/logout invalidates the current token immediately

Scheduling & Reliability

• APScheduler with PostgreSQL jobstore — scheduled jobs survive restarts
• Visual cron builder: frequency presets, live expression preview, next-5-runs display
• Hot reload: schedule changes sync to the scheduler automatically every 60 seconds
• Celery / Redis task queue (optional): when FLOWFORGE_REDIS_URL is set, pipeline runs are dispatched as Celery tasks; falls back to background threads without Redis
• Distributed concurrency lock: Redis-backed run-limit lock so multiple API/worker instances can be scaled horizontally without over-running a pipeline
• Graceful shutdown: SIGTERM handler drains in-flight runs before exit

Observability & Production Ops

• Prometheus metrics: GET /api/metrics exposes run totals, active runs, and queue depth in plain-text exposition format
• Celery Flower dashboard: optional --profile monitoring Docker Compose service for worker/task visibility
• Gunicorn + Nginx deployment guide: gevent workers, systemd units, TLS termination — see docs/deployment.md / RUNBOOK.md
• SQLAlchemy pool tuning: pool size/overflow/timeout/recycle configurable via env vars, with PgBouncer guidance for 100+ concurrent pipelines

AI Features (Ollama, Claude API, or Gemini API)

All AI-powered UI features (below) run through Ollama by default (local, zero cost) and fall back to Claude or Gemini — in that order — if Ollama is unreachable and one of ANTHROPIC_API_KEY / GEMINI_API_KEY is set. Configure in Settings → AI.

• AI chart generator: "Visualize" button on Report Preview — sends column names + sample rows to the AI provider; renders result with Recharts (bar, line, area, pie, scatter)
• SQL Explainer: "Explain" button in the SQL editor — plain-English summary of what a query does
• SQL Optimizer: "Optimize" button — side-by-side diff of original vs AI-suggested rewrite; accept with one click
• Pipeline failure diagnosis: "Explain this error" on failed step logs — 2–4 sentence cause and fix
• Data Profiler: one-click narrative summary of query result (structure, ranges, nulls, outliers)
• Run history anomaly alerts: statistical outlier detection (>2σ) on row counts and durations; AI narrative when anomaly is detected

Security

• AES-256-GCM credential encryption for all DB and email provider configs stored in the database
• Report encryption at rest — optional (FLOWFORGE_ENCRYPT_OUTPUT=true); files decrypted transparently on download
• bcrypt admin password hashing; separate JWT signing secret from encryption key
• GDPR data export & purge — admin endpoints to export a user's profile/audit/run history, or anonymise-and-delete
• Login rate limiting (10/min per IP) and manual trigger rate limiting
• Jinja2 SandboxedEnvironment — no arbitrary code execution in templates
• Env var allowlist (FLOWFORGE_TEMPLATE_ENV_VARS) — restrict which env vars are accessible in templates
• Path containment check on report file download — prevents directory traversal
• Procedure name validation — blocks SQL injection via stored procedure names
• SSRF guard on outbound webhook/notification URLs — rejects internal/private-network targets
• SFTP strict host key checking (FLOWFORGE_SFTP_STRICT_HOSTKEYS=true)
• Audit log — every login, pipeline run, config change, and webhook trigger logged with username attribution; file (rotating 10 MB × 5) + structured JSON stdout
• SAST in CI — bandit (Python) + npm audit on every push; TruffleHog secrets scan; property-based fuzz tests (Hypothesis)
• Signed releases — SLSA build provenance attestation on tagged releases; PyPI publish via OIDC trusted publishing

Developer Experience

• In-app help: context-sensitive help drawer, page intro cards, field tooltips, concept glossary, FAQ link
• Webhook / API trigger: POST /pipelines/{id}/trigger?token=... for external integration; per-pipeline tokens with audit trail
• Output cleanup: CLI command + daily scheduler job to prune old report files by TTL
• Docker Compose: one-command local stack (API + frontend + PostgreSQL + scheduler + Redis + Celery worker)
• GitHub Actions CI: full pytest suite + SAST + frontend audit on every push and PR

Comparison

Step Types

Custom step types, database connections, and email providers can also be added via the plugin system without forking FlowForge.

Supported Technologies

Quick Start

Option A — Docker Compose (recommended)

git clone https://github.com/jagdeepvirdi/flowforge.git
cd flowforge
cp .env.example .env
# Edit .env — set FLOWFORGE_SECRET_KEY and FLOWFORGE_PASSWORD at minimum
docker compose up

Open http://localhost:5000. The stack starts the API, frontend (Nginx), PostgreSQL, scheduler, Redis, and a Celery worker.

For Oracle support:

docker compose -f docker-compose.yml -f docker-compose.oracle.yml up

Option B — Local Development

1. Prerequisites

• Python 3.11+
• Node.js 18+
• PostgreSQL (for FlowForge's own config database)

2. Install

pip install -e .
# Optional extras:
pip install -e ".[pdf]"           # PDF report generation (WeasyPrint)
pip install -e ".[oracle]"        # Oracle database support (python-oracledb)
pip install -e ".[gmail]"         # Gmail email provider + Google Drive
pip install -e ".[microsoft365]"  # Microsoft 365 email + OneDrive
pip install -e ".[mysql]"         # MySQL / MariaDB support
pip install -e ".[mssql]"         # MSSQL / SQL Server support (pyodbc)
pip install -e ".[snowflake]"     # Snowflake connector
pip install -e ".[bigquery]"      # BigQuery connector
pip install -e ".[ses]"           # AWS SES email provider + S3 upload
pip install -e ".[azure_blob]"    # Azure Blob Storage upload
pip install -e ".[sftp]"          # SFTP transfer step (paramiko)
pip install -e ".[sso]"           # SSO: Google, Microsoft, SAML 2.0
pip install -e ".[claude]"        # ai_analyze step via Claude API
pip install -e ".[all]"           # Everything above

ai_analyze step via Gemini API needs no extra install — it's a plain REST call, just set GEMINI_API_KEY.

3. Configure

cp .env.example .env
# Set FLOWFORGE_DB_URL, FLOWFORGE_SECRET_KEY, and FLOWFORGE_PASSWORD at minimum

4. Initialize the database and create the admin user

flowforge db upgrade
flowforge db seed

5. Start

Windows:

.\flowforge.ps1 dev start

macOS / Linux:

./flowforge.sh dev start

Both scripts start the Flask API, APScheduler, and Vite frontend together. Open http://localhost:5173.

Other actions: stop, restart, and status (e.g. .\flowforge.ps1 dev status).

6. Production mode (serves built frontend from Flask)

.\flowforge.ps1 prod start   # Windows
./flowforge.sh prod start    # macOS/Linux

7. With Celery task queue (optional)

# Start Redis, then run a worker alongside the web server:
flowforge worker --concurrency 4
# Or: celery -A flowforge.celery_app worker --concurrency 4 --loglevel info

When FLOWFORGE_REDIS_URL is set in .env, pipeline runs are dispatched to the Celery worker automatically. Without it, runs execute in background threads.

CLI Reference

# Server
flowforge web                           # start Flask dev server
flowforge schedule                      # start APScheduler daemon
flowforge worker [--concurrency N]      # start Celery worker (requires FLOWFORGE_REDIS_URL)

# Pipelines
flowforge run "Monthly Revenue Report"  # run a pipeline by name
flowforge list                          # list all pipelines with schedule and status
flowforge validate "Pipeline Name"      # validate pipeline config and connections
flowforge export "Pipeline Name"        # export pipeline as YAML
flowforge import pipeline.yaml          # import pipeline from YAML (--overwrite to replace)

# Database
flowforge db upgrade [revision]         # apply Alembic migrations (default: head)
flowforge db downgrade <revision>       # revert migrations
flowforge db current                    # show current migration revision
flowforge db stamp <revision>           # mark DB at revision without running migrations
flowforge db seed                       # create admin user from env vars (run once after upgrade)

# Maintenance
flowforge cleanup [--days N] [--dry-run]  # remove output files older than N days

# OAuth2 setup wizards
flowforge setup gmail
flowforge setup microsoft365

Variable System

Available in every config field rendered via Jinja2:

Multi-User Roles

Role is set per user by an admin. The UI automatically hides write actions for viewers. Users can also sign in via MFA (TOTP), SSO (Google, Microsoft, or SAML 2.0), or reset a forgotten password via emailed link — see the Security features above.

API Trigger (Webhook)

Pipelines can be triggered externally without a UI login:

# Create a token in the UI (Settings → Pipeline → Webhook Tokens)
# Then trigger via HTTP:
curl -X POST "https://your-flowforge/api/pipelines/{id}/trigger?token=flwf_..."

Returns { run_id, status, pipeline_name } with HTTP 202. The token is audited on every use.

Project Structure

flowforge/
├── flowforge/               # Python package
│   ├── cli.py               # Click CLI
│   ├── celery_app.py        # Celery instance + Flask context wiring
│   ├── tasks.py             # Celery task definitions
│   ├── audit.py             # Audit log (file + JSON stdout)
│   ├── engine/              # runner, scheduler, launcher, loader (+ plugins), context, shutdown
│   ├── steps/               # db_procedure, db_query, report, email, drive_upload, onedrive_upload,
│   │                        #   s3_upload, azure_blob_upload, data_load, bulk_load, sftp_transfer,
│   │                        #   ssh_command, ssh_health_check, db_health_check, script_report,
│   │                        #   notification, ai_analyze
│   ├── connections/         # PostgreSQL, Oracle, MySQL, MSSQL, ODBC, Snowflake, BigQuery, Redshift + factory
│   ├── email_providers/     # Gmail, M365, SMTP, SendGrid, SES, Mailgun + factory
│   ├── reports/             # Excel, CSV, PDF, JSON generators
│   ├── storage/             # Google Drive, OneDrive, S3, Azure Blob clients
│   ├── crypto.py            # AES-256-GCM credential + file encryption
│   ├── net_guard.py         # SSRF guard for outbound webhook/notification URLs
│   ├── db/                  # SQLAlchemy models + Alembic migrations
│   └── api/                 # Flask REST API (routes, auth, mfa, sso, validators, serializers)
├── frontend/                # React + Vite + TypeScript
│   └── src/
│       ├── pages/           # Dashboard, Pipelines, Reports, Emails, Connections, Recipients,
│       │                    #   RunHistory, Settings, Users, BulkLoads, Projects, Login, AuditLog
│       ├── components/      # pipeline/, email/, report/, runs/, shared/
│       └── lib/             # api.ts, auth.ts, types.ts, store.ts
├── docs/                    # Guides and runbook
├── tests/                   # pytest suite (1,800+ tests)
├── .github/workflows/       # GitHub Actions CI
├── docker-compose.yml       # API + frontend + PostgreSQL + scheduler + Redis + worker
├── docker-compose.oracle.yml  # Oracle 23c overlay
├── flowforge.ps1            # Windows dev/prod startup script
├── flowforge.sh             # macOS/Linux dev/prod startup script
├── pyproject.toml
├── .env.example
└── CHANGELOG.md

Documentation

• docs/getting-started.md — end-to-end setup walkthrough
• docs/step-types.md — full config reference for all step types
• docs/email-providers.md — SMTP, M365, Gmail, SendGrid, SES, and Mailgun setup guides
• docs/plugins.md — writing a custom plugin step type, database connection, or email provider
• docs/deployment.md — production deployment (Docker Compose or Gunicorn + Nginx + systemd)
• docs/security.md / docs/threat-model.md / docs/data-flow.md — security model, threat model, and compliance reference
• docs/RUNBOOK.md — database migrations, startup sequence, production ops
• CONTRIBUTING.md — dev setup, running tests, PR process
• CHANGELOG.md — version history
• ROADMAP.md / docs/TASKS.md — current status and active work (source of truth over the table below)

Roadmap

ROADMAP.md and docs/TASKS.md are the continuously-updated source of truth for what's shipped — treat the table below as a summary snapshot, not the latest word.

Contributing

Obtain: Install via Docker Compose (docker compose up) or locally — see Quick Start above.

Feedback: Use GitHub Issues to report bugs or request features. Please include your FlowForge version, Python version, OS, steps to reproduce, and any relevant log output from ff_step_runs.logs.

Contribute: See CONTRIBUTING.md for dev setup, running tests, adding step types, and the PR process. All contributions welcome — bug fixes, new step types, documentation improvements, and test coverage.

License

MIT — see LICENSE.