Pipekit

A database sync tool with a web UI. A module defines one job — a source query against one database, merged into a destination table on another. Pipekit resolves watermarks, runs the extract, stages the rows, merges them, and records what happened. Modules can be grouped, ordered, and scheduled with cron.

It is an orchestration layer over jrunner, an external Java CLI that does all the actual JDBC work. Pipekit never talks to a remote database directly — it shells out. What it adds is the state, history, and UI that a directory of shell scripts can't provide: run logs with timings and row counts, live output while a job runs, a wizard for authoring new modules, and a scheduler.

---

Install

Requirements

• Python 3.10+
• jrunner on PATH, with JAVA_HOME set — the Java CLI that performs every JDBC read and write. Install /opt/jrunner first; deploy.sh refuses to run without it and injects the detected JAVA_HOME into the systemd unit.
• sudo — the deploy creates a system user, writes to /etc/pipekit, and installs a systemd unit.

Deploy

./deploy.sh

Idempotent — re-run it after any code update. It creates the pipekit system user and group, adds you to that group, builds the venv at .venv, installs the /usr/local/bin/pipekit launcher, creates /etc/pipekit/secrets.env (mode 0640), initialises the SQLite schema, registers a JDBC driver row for every jar shipped with jrunner, and installs the systemd unit. It does not start the service.

Then add your connection passwords and start it:

sudo pipekit secrets set MYDB_PW          # prompts if the value is omitted
sudo systemctl start pipekit

The web UI is on api_port from config.yaml — 8200 by default.

Service management

sudo systemctl status pipekit
sudo systemctl restart pipekit      # after any code change
sudo journalctl -u pipekit -f       # follow logs

Running outside the service

pipekit serve                                        # host/port from config.yaml
pipekit serve --host 0.0.0.0 --port 8080 --reload    # dev, auto-reload

CLI

pipekit init                    # create/upgrade the SQLite schema
pipekit doctor                  # health check: config, jrunner, DB
pipekit run <module_name>       # run one module synchronously
pipekit set-password <user>     # HTTP Basic Auth credentials for /api/*
pipekit secrets set KEY [VALUE] # add/update a secret
pipekit secrets list            # show stored keys, never values
pipekit drivers list            # registered driver kinds

Configuration also round-trips to plain text, so modules can be diffed and version-controlled rather than living only in the database:

pipekit export                  # dump drivers/connections/modules/groups to <repo>/config
pipekit apply --dry-run         # print the change plan without touching the DB
pipekit apply                   # rehydrate from those files, matching by name
pipekit apply --prune           # also delete rows absent from the files

Configuration

config.yaml in the repo root holds the database path, jrunner path, driver directory, API host/port, and an optional SMTP block. Two environment variables override the defaults: PIPEKIT_CONFIG for the config file and PIPEKIT_SECRETS for the secrets file.

---

About the project

Tech stack

• Python 3.10+ — no ORM, no migrations framework, no task queue
• FastAPI + Uvicorn for both the JSON API and the HTML pages
• Jinja2 templates; HTMX (from CDN) plus vanilla JS for interactivity. There is no build step and no frontend toolchain. Alpine.js is named in some older docs but is not loaded.
• SQLite (stdlib sqlite3) — a single pipekit.db file holds all state
• PyYAML for config, croniter for schedule evaluation
• python-multipart — required for HTML form POSTs; it is not a transitive dependency of the FastAPI base wheel, and form posts return 500 without it
• jrunner — a separate Java tool, the only thing that speaks JDBC

Deployed as the pipekit system user, venv at /opt/pipekit/.venv, secrets at /etc/pipekit/secrets.env.

Architecture

Five layers, bottom to top:

1. SQLite — one file, all state
2. repo.py — every CRUD operation for every table; the only module that touches the database
3. engine/ — runs a module: acquires the lock, resolves watermarks, calls jrunner, builds and runs the merge, fires hooks, writes the run log, releases the lock
4. api/ — JSON REST under /api/*, HTTP Basic Auth (except /health)
5. web/ — HTML pages at /, currently unauthenticated

Data model

• driver — a registered JDBC driver (kind, jar, class, URL template)
• connection — a named database (JDBC URL, username, password)
• module — one sync job: source and dest connection, source query, merge strategy and key, column map, enabled flag, running lock
• watermark — a named placeholder with its own resolver SQL. The first cell of the first row is substituted for {watermark_name} in the source query.
• hook — post-merge SQL, ordered, fired on success / failure / always
• run_log — immutable history: resolved SQL, merge SQL, watermark values, stdout/stderr, timings, and the live log streamed during the run
• grp / group_member / group_run — named, ordered sets of modules and their executions
• schedule — a cron expression bound to a group, evaluated in local server time

How a run works

run_module(module_id)
  → atomic UPDATE module SET running=1 WHERE running=0   # fails if already locked
  → resolve each watermark via jrunner                   # always live, even on dry runs
  → substitute {name} tokens into the source query
  → build the merge SQL
  → if dry_run: write the run_log and stop — no data moves
  → DROP + CREATE pipekit_staging.{module_name}          # self-heals schema drift
  → jrunner migrate: source → staging                    # stdout streamed to live_log
  → run the merge, then the hooks in order
  → write run_log; UPDATE module SET running=0 (in finally)

A group run creates a group_run row and calls run_module for each enabled member in order. It continues past individual failures — every member runs — and the group's final status reflects the worst outcome.

Run statuses: running, success, dry_run, error, cancelled.

Merge strategies

• full — TRUNCATE the dest, then insert everything
• incremental — delete the rows matching merge_key, then insert. The key is stored comma-separated and expands to a multi-column predicate.
• append — insert only

The staging table is recreated on every run and exists only for the duration of that run.

Credentials

Connection passwords are stored as $VAR_NAME references, never as values. They resolve at run time from /etc/pipekit/secrets.env. Secrets never reach the database, and never appear on a command line.

Performance note

When the destination is SQL Server or Postgres, jrunner is invoked with -b so the load uses the native bulk path — SQLServerBulkCopy over TDS, or COPY … FROM STDIN. This is automatic, per-destination, with nothing to configure, and the difference is large: one 1.27M-row SQL Server load went from ~111 minutes to ~4. DB2 destinations keep the ordinary INSERT path.

Reconciliation

reconcile.py compares a module's live source against its synced destination using column-wise aggregates — counts, sums, min/max, summed lengths. These rely only on arithmetic and ordering, which every backend computes identically, so no shared hash function or byte-identical row serialisation is needed.

PIPEKIT_SECRETS=/etc/pipekit/secrets.env .venv/bin/python reconcile.py <module> [--quick]

Three depths: the default compares every column; --quick drops min/max and length sums; --super-quick compares only row counts, which takes seconds even on millions of rows and still catches missing or duplicated rows. Exit status is non-zero when anything diverges. Note that it compares whole tables against a live source, so an incremental module will always show drift proportional to the time since its last run.

---

Further reading

• SPEC.md — the authoritative design document, and the rationale behind every architectural decision here
• CLAUDE.md — orientation for AI coding agents working in this repo