Paul Trowbridge
dd9f89f4b5
Add `pipekit export` / `pipekit apply` (pipekit/config_io.py) to serialise the DB's config — drivers, connections, modules (+ columns, watermarks, hooks), groups, schedules — to a git-trackable text tree under config/, and rehydrate it. SQLite stays runtime state; definitions become diffable/reviewable/ revertible. - config only: run_log/group_run/settings and per-run state columns excluded - name-keyed refs (portable across databases); source_query in .sql sidecars; columns as real JSON arrays for line-by-line diffs - newline-normalised so CRLF-vs-LF is never a spurious change - apply is create/update by name; child collections fully synced; top-level deletes gated behind --prune; --dry-run prints the plan - round-trip is identity (export -> apply --dry-run == nothing to do) Commits the current config/ as the first baseline, capturing the freshly populated columns_json for rm00101/rm00301/iv00101. Documented in SPEC.md. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
339 lines
15 KiB
Python
339 lines
15 KiB
Python
"""Export / apply the pipekit *configuration* as version-controllable text.
|
|
|
|
The SQLite database is runtime state; the pipeline *definitions* it holds
|
|
(drivers, connections, modules, watermarks, hooks, groups, schedules) are
|
|
config and belong in git. This module serialises those definitions to a tree
|
|
of text files and rehydrates them, so changes get diff / review / rollback.
|
|
|
|
Layout (under the config dir, default ``<repo>/config``)::
|
|
|
|
drivers.json # driver registry
|
|
connections.json # connections (passwords are $ENV refs, not secrets)
|
|
groups.json # groups + members + schedules
|
|
modules/<name>.json # one module's definition (+ watermarks, hooks)
|
|
modules/<name>.sql # that module's source_query (sidecar, clean diffs)
|
|
|
|
Everything is keyed by natural name (not autoincrement id) so files stay
|
|
portable across databases. Run-history tables (run_log, group_run, settings)
|
|
and per-run state columns (running, next_resolved_query, timestamps) are
|
|
deliberately excluded.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from . import repo
|
|
|
|
|
|
def default_dir() -> Path:
|
|
return Path(__file__).resolve().parent.parent / "config"
|
|
|
|
|
|
def _write_json(path: Path, obj) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(obj, indent=2, ensure_ascii=False) + "\n")
|
|
|
|
|
|
def _read_json(path: Path):
|
|
return json.loads(path.read_text())
|
|
|
|
|
|
def _norm_sql(s: str | None) -> str:
|
|
"""Canonical form for comparing SQL: LF newlines, no trailing blank lines.
|
|
Line-ending style (CRLF vs LF) isn't meaningful and must not read as a diff."""
|
|
return (s or "").replace("\r\n", "\n").replace("\r", "\n").rstrip("\n")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Export
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def export_config(dest: Path) -> dict:
|
|
"""Write the full config tree under ``dest``. Returns a summary count."""
|
|
dest = Path(dest)
|
|
conn_by_id = {c["id"]: c["name"] for c in repo.list_connections()}
|
|
drv_by_id = {d["id"]: d["name"] for d in repo.list_drivers()}
|
|
|
|
# drivers
|
|
drivers = [
|
|
{"name": d["name"], "kind": d["kind"], "jar_file": d["jar_file"],
|
|
"class_name": d["class_name"], "url_template": d["url_template"]}
|
|
for d in repo.list_drivers()
|
|
]
|
|
_write_json(dest / "drivers.json", drivers)
|
|
|
|
# connections (name-keyed refs; password is an env-var reference)
|
|
connections = [
|
|
{"name": c["name"], "driver": drv_by_id.get(c["driver_id"]),
|
|
"jdbc_url": c["jdbc_url"], "username": c["username"],
|
|
"password": c["password"],
|
|
"default_dest_connection": conn_by_id.get(c["default_dest_connection_id"]),
|
|
"default_dest_schema": c["default_dest_schema"], "notes": c["notes"]}
|
|
for c in repo.list_connections()
|
|
]
|
|
_write_json(dest / "connections.json", connections)
|
|
|
|
# modules — one file per module, SQL split into a sidecar
|
|
mod_dir = dest / "modules"
|
|
mod_dir.mkdir(parents=True, exist_ok=True)
|
|
kept = set()
|
|
for m in repo.list_modules():
|
|
full = repo.get_module(m["id"])
|
|
cols = json.loads(full["columns_json"] or "[]")
|
|
wms = [
|
|
{"name": w["name"], "connection": conn_by_id.get(w["connection_id"]),
|
|
"resolver_sql": w["resolver_sql"], "default_value": w["default_value"]}
|
|
for w in repo.list_watermarks(full["id"])
|
|
]
|
|
hooks = [
|
|
{"run_order": h["run_order"],
|
|
"connection": conn_by_id.get(h["connection_id"]),
|
|
"sql": h["sql"], "run_on": h["run_on"]}
|
|
for h in repo.list_hooks(full["id"])
|
|
]
|
|
record = {
|
|
"name": full["name"],
|
|
"source_connection": conn_by_id.get(full["source_connection_id"]),
|
|
"dest_connection": conn_by_id.get(full["dest_connection_id"]),
|
|
"dest_table": full["dest_table"],
|
|
"staging_table": full["staging_table"],
|
|
"merge_strategy": full["merge_strategy"],
|
|
"merge_key": full["merge_key"],
|
|
"enabled": full["enabled"],
|
|
"dest_description": full["dest_description"],
|
|
"columns": cols, # real array -> per-column line diffs
|
|
"watermarks": wms,
|
|
"hooks": hooks,
|
|
}
|
|
_write_json(mod_dir / f"{full['name']}.json", record)
|
|
(mod_dir / f"{full['name']}.sql").write_text(
|
|
_norm_sql(full["source_query"]) + "\n")
|
|
kept.add(full["name"])
|
|
|
|
# prune stale module files no longer in the DB
|
|
for f in mod_dir.glob("*.json"):
|
|
if f.stem not in kept:
|
|
f.unlink()
|
|
f.with_suffix(".sql").unlink(missing_ok=True)
|
|
|
|
# groups (+ members + schedules)
|
|
groups = []
|
|
for g in repo.list_groups():
|
|
members = [
|
|
{"module": mm["module_name"], "run_order": mm["run_order"]}
|
|
for mm in repo.list_group_members(g["id"])
|
|
]
|
|
scheds = [
|
|
{"cron_expr": s["cron_expr"], "enabled": s["enabled"]}
|
|
for s in repo.list_schedules_for_group(g["id"])
|
|
]
|
|
groups.append({"name": g["name"], "members": members, "schedules": scheds})
|
|
_write_json(dest / "groups.json", groups)
|
|
|
|
return {"drivers": len(drivers), "connections": len(connections),
|
|
"modules": len(kept), "groups": len(groups), "dir": str(dest)}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Apply
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class ApplyError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def _plan(actions: list, verb: str, kind: str, name: str) -> None:
|
|
actions.append(f"{verb:7} {kind:11} {name}")
|
|
|
|
|
|
def apply_config(src: Path, *, dry_run: bool = False, prune: bool = False) -> list[str]:
|
|
"""Rehydrate the DB from the config tree under ``src``.
|
|
|
|
Create/update by name; child collections (watermarks, hooks, schedules,
|
|
group members) are fully synced to match the files. Top-level rows present
|
|
in the DB but absent from the files are only removed with ``prune=True``.
|
|
Returns the list of planned/performed actions.
|
|
"""
|
|
src = Path(src)
|
|
if not (src / "connections.json").exists():
|
|
raise ApplyError(f"no config found at {src} (expected connections.json)")
|
|
actions: list[str] = []
|
|
|
|
# --- drivers ---
|
|
for d in _read_json(src / "drivers.json"):
|
|
cur = repo.get_driver_by_name(d["name"])
|
|
if cur is None:
|
|
_plan(actions, "create", "driver", d["name"])
|
|
if not dry_run:
|
|
repo.create_driver(name=d["name"], kind=d["kind"],
|
|
jar_file=d["jar_file"], class_name=d["class_name"],
|
|
url_template=d.get("url_template"))
|
|
else:
|
|
changed = any(cur.get(k) != d.get(k) for k in
|
|
("jar_file", "class_name", "url_template"))
|
|
if changed:
|
|
_plan(actions, "update", "driver", d["name"])
|
|
if not dry_run:
|
|
repo.update_driver(cur["id"], jar_file=d["jar_file"],
|
|
class_name=d["class_name"],
|
|
url_template=d.get("url_template"))
|
|
|
|
# --- connections (two passes: create/update, then wire default-dest) ---
|
|
conns = _read_json(src / "connections.json")
|
|
for c in conns:
|
|
drv = repo.get_driver_by_name(c["driver"])
|
|
if drv is None:
|
|
raise ApplyError(f"connection {c['name']}: unknown driver {c['driver']!r}")
|
|
cur = repo.get_connection_by_name(c["name"])
|
|
fields = dict(driver_id=drv["id"], jdbc_url=c["jdbc_url"],
|
|
username=c.get("username"), password=c.get("password"),
|
|
default_dest_schema=c.get("default_dest_schema"),
|
|
notes=c.get("notes"))
|
|
if cur is None:
|
|
_plan(actions, "create", "connection", c["name"])
|
|
if not dry_run:
|
|
repo.create_connection(name=c["name"], **fields)
|
|
else:
|
|
watch = ("jdbc_url", "username", "password", "default_dest_schema", "notes")
|
|
if drv["id"] != cur["driver_id"] or any(cur.get(k) != c.get(k) for k in watch):
|
|
_plan(actions, "update", "connection", c["name"])
|
|
if not dry_run:
|
|
repo.update_connection(cur["id"], **fields)
|
|
# pass 2: default_dest_connection_id (now every connection exists)
|
|
if not dry_run:
|
|
for c in conns:
|
|
dd = c.get("default_dest_connection")
|
|
if not dd:
|
|
continue
|
|
me = repo.get_connection_by_name(c["name"])
|
|
target = repo.get_connection_by_name(dd)
|
|
if me and target and me.get("default_dest_connection_id") != target["id"]:
|
|
repo.update_connection(me["id"], default_dest_connection_id=target["id"])
|
|
|
|
# --- modules ---
|
|
mod_dir = src / "modules"
|
|
file_modules = set()
|
|
for jf in sorted(mod_dir.glob("*.json")):
|
|
rec = _read_json(jf)
|
|
name = rec["name"]
|
|
file_modules.add(name)
|
|
sqlf = jf.with_suffix(".sql")
|
|
source_query = _norm_sql(sqlf.read_text()) if sqlf.exists() else ""
|
|
sconn = repo.get_connection_by_name(rec["source_connection"])
|
|
dconn = repo.get_connection_by_name(rec["dest_connection"])
|
|
if not sconn or not dconn:
|
|
raise ApplyError(f"module {name}: unknown connection "
|
|
f"(source={rec['source_connection']} dest={rec['dest_connection']})")
|
|
cur = repo.get_module_by_name(name)
|
|
cols = rec.get("columns") or []
|
|
if cur is None:
|
|
_plan(actions, "create", "module", name)
|
|
if not dry_run:
|
|
mod = repo.create_module(
|
|
name=name, source_connection_id=sconn["id"],
|
|
dest_connection_id=dconn["id"], dest_table=rec["dest_table"],
|
|
source_query=source_query, merge_strategy=rec["merge_strategy"],
|
|
merge_key=rec.get("merge_key"),
|
|
staging_table=rec.get("staging_table"),
|
|
columns=cols or None, dest_description=rec.get("dest_description"))
|
|
mid = mod["id"]
|
|
else:
|
|
mid = None
|
|
else:
|
|
mid = cur["id"]
|
|
cur_cols = json.loads(cur["columns_json"] or "[]")
|
|
watch = ("dest_table", "staging_table", "merge_strategy", "merge_key",
|
|
"dest_description", "enabled")
|
|
meta_changed = (sconn["id"] != cur["source_connection_id"]
|
|
or dconn["id"] != cur["dest_connection_id"]
|
|
or source_query != _norm_sql(cur["source_query"])
|
|
or any(cur.get(k) != rec.get(k) for k in watch))
|
|
cols_changed = cols != cur_cols
|
|
if meta_changed or cols_changed:
|
|
_plan(actions, "update", "module", name
|
|
+ ("" if meta_changed else " (columns)"))
|
|
if not dry_run:
|
|
repo.update_module(
|
|
mid, source_connection_id=sconn["id"],
|
|
dest_connection_id=dconn["id"], dest_table=rec["dest_table"],
|
|
staging_table=rec.get("staging_table"),
|
|
source_query=source_query, merge_strategy=rec["merge_strategy"],
|
|
merge_key=rec.get("merge_key"),
|
|
dest_description=rec.get("dest_description"),
|
|
enabled=rec.get("enabled"))
|
|
if cols_changed:
|
|
repo.update_module_columns(mid, cols)
|
|
|
|
# child collections: watermarks (keyed by name) + hooks (rebuilt)
|
|
if not dry_run and mid is not None:
|
|
want_wms = {w["name"]: w for w in rec.get("watermarks", [])}
|
|
have_wms = {w["name"]: w for w in repo.list_watermarks(mid)}
|
|
for wname, w in want_wms.items():
|
|
wc = repo.get_connection_by_name(w["connection"]) if w.get("connection") else None
|
|
cid = wc["id"] if wc else None
|
|
if wname in have_wms:
|
|
repo.update_watermark(have_wms[wname]["id"], connection_id=cid,
|
|
resolver_sql=w["resolver_sql"],
|
|
default_value=w.get("default_value"))
|
|
else:
|
|
repo.create_watermark(module_id=mid, name=wname, connection_id=cid,
|
|
resolver_sql=w["resolver_sql"],
|
|
default_value=w.get("default_value"))
|
|
for wname, w in have_wms.items():
|
|
if wname not in want_wms:
|
|
repo.delete_watermark(w["id"])
|
|
# hooks have no natural key -> rebuild from file
|
|
for h in repo.list_hooks(mid):
|
|
repo.delete_hook(h["id"])
|
|
for h in rec.get("hooks", []):
|
|
hc = repo.get_connection_by_name(h["connection"]) if h.get("connection") else None
|
|
repo.create_hook(module_id=mid, sql=h["sql"],
|
|
run_order=h.get("run_order", 0),
|
|
connection_id=hc["id"] if hc else None,
|
|
run_on=h.get("run_on", "success"))
|
|
|
|
# prune modules absent from files
|
|
for m in repo.list_modules():
|
|
if m["name"] not in file_modules:
|
|
if prune:
|
|
_plan(actions, "delete", "module", m["name"])
|
|
if not dry_run:
|
|
repo.delete_module(m["id"])
|
|
else:
|
|
_plan(actions, "skip", "module", m["name"] + " (in db, not in files)")
|
|
|
|
# --- groups (+ members + schedules) ---
|
|
file_groups = set()
|
|
for g in _read_json(src / "groups.json"):
|
|
file_groups.add(g["name"])
|
|
cur = repo.get_group_by_name(g["name"])
|
|
if cur is None:
|
|
_plan(actions, "create", "group", g["name"])
|
|
if not dry_run:
|
|
cur = repo.create_group(name=g["name"])
|
|
if not dry_run and cur is not None:
|
|
members = []
|
|
for mm in g.get("members", []):
|
|
mod = repo.get_module_by_name(mm["module"])
|
|
if mod:
|
|
members.append({"module_id": mod["id"],
|
|
"run_order": mm.get("run_order", 0)})
|
|
repo.set_group_members(cur["id"], members)
|
|
for s in repo.list_schedules_for_group(cur["id"]):
|
|
repo.delete_schedule(s["id"])
|
|
for s in g.get("schedules", []):
|
|
repo.create_schedule(group_id=cur["id"], cron_expr=s["cron_expr"],
|
|
enabled=s.get("enabled", 1))
|
|
|
|
for g in repo.list_groups():
|
|
if g["name"] not in file_groups:
|
|
if prune:
|
|
_plan(actions, "delete", "group", g["name"])
|
|
if not dry_run:
|
|
repo.delete_group(g["id"])
|
|
else:
|
|
_plan(actions, "skip", "group", g["name"] + " (in db, not in files)")
|
|
|
|
return actions
|