Paul Trowbridge
72750cb57f
The engine built staging as `LIKE dest` and merged into it without ever
checking the destination existed — a missing dest (e.g. a file-based deploy
that never went through the web wizard) failed deep in the staging step with
a raw `relation ... does not exist`.
Add `engine/dest.reconcile_dest()`, called as step 3b of run_module (after the
dry-run return, before staging):
- missing + not create -> DestError telling the user to pass --create-dest
- missing + create -> CREATE SCHEMA + CREATE TABLE from the module's
column map
- exists -> fail loudly if any mapped column is absent (drift);
no auto-ALTER, since the positional load can't
tolerate a mid-table add
Expose via `pipekit run --create-dest`. Default (scheduled groups, API) is
unchanged: validate-and-fail, never silently provision.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
86 lines
3.3 KiB
Python
86 lines
3.3 KiB
Python
"""Destination-table preflight.
|
|
|
|
Ensure a module's destination table exists and covers the module's column map
|
|
*before* a run builds staging as ``LIKE dest`` and merges into it. Without this
|
|
the engine would fail deep inside the staging step with a raw
|
|
``relation ... does not exist`` when the dest was never provisioned (e.g. a
|
|
file-based deploy that never went through the wizard).
|
|
|
|
Kept driver-agnostic — all DDL/introspection goes through the dest
|
|
:class:`~pipekit.drivers.base.Driver`.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from .. import jrunner
|
|
|
|
|
|
class DestError(RuntimeError):
|
|
"""Dest table is missing (and create not requested) or has drifted."""
|
|
|
|
|
|
def _split(dest_table: str) -> tuple[str, str]:
|
|
"""Split ``schema.table`` → (schema, table); default schema is ``public``."""
|
|
schema, _, bare = dest_table.partition(".")
|
|
if not bare:
|
|
return "public", schema
|
|
return schema, bare
|
|
|
|
|
|
def module_columns(module: dict) -> list[dict]:
|
|
"""The module's stored column map, or [] if none recorded."""
|
|
raw = module.get("columns_json")
|
|
return json.loads(raw) if raw else []
|
|
|
|
|
|
def reconcile_dest(module: dict, dest_drv, dest_conn: dict, *,
|
|
create: bool) -> str:
|
|
"""Ensure ``module['dest_table']`` exists on ``dest_conn`` and covers the
|
|
module's dest columns. Returns a short action string for the run log.
|
|
|
|
* missing + ``create`` → CREATE SCHEMA (if needed) + CREATE TABLE from
|
|
the module's column map.
|
|
* missing + not ``create`` → raise :class:`DestError`.
|
|
* exists → raise :class:`DestError` if any dest column is
|
|
absent (drift); otherwise a no-op.
|
|
"""
|
|
if dest_drv is None:
|
|
return "skipped (no dest driver)"
|
|
|
|
schema, bare = _split(module["dest_table"])
|
|
try:
|
|
existing = dest_drv.check_dest_table(dest_conn, schema, bare)
|
|
except jrunner.JrunnerError as e:
|
|
raise DestError(
|
|
f"could not introspect dest {module['dest_table']}: {e}") from e
|
|
|
|
cols = module_columns(module)
|
|
|
|
if existing is None:
|
|
if not create:
|
|
raise DestError(
|
|
f"dest table {module['dest_table']} does not exist — re-run "
|
|
"with --create-dest to provision it from the module's columns")
|
|
if not cols:
|
|
raise DestError(
|
|
f"cannot create {module['dest_table']}: module has no column map")
|
|
qualified = dest_drv.qualified_table_name(bare, schema=schema)
|
|
if schema:
|
|
jrunner.run_dest_sql(dest_conn, dest_drv.create_schema_sql(schema))
|
|
jrunner.run_dest_sql(
|
|
dest_conn, dest_drv.build_create_table_sql(qualified, cols))
|
|
return f"created {module['dest_table']} ({len(cols)} columns)"
|
|
|
|
# Exists — verify its shape covers every column we intend to load. The
|
|
# merge is a positional `SELECT *`, so a dest missing a column would either
|
|
# error or silently misalign; fail loudly instead.
|
|
missing = [c["dest_name"] for c in cols
|
|
if c["dest_name"].lower() not in existing]
|
|
if missing:
|
|
raise DestError(
|
|
f"dest table {module['dest_table']} is missing column(s): "
|
|
f"{', '.join(missing)} — reconcile the schema before running")
|
|
return f"verified {module['dest_table']}"
|