"""Apply schema.sql — plan, apply, then VERIFY.
Three platform facts shape this file, and the third one is why it does more than run DDL:
- `CREATE TABLE IF NOT EXISTS` is refused when the table exists with a DIFFERENT shape, so
new columns go at the END of a CREATE and are also added as explicit ALTERs below it.
- `ALTER TABLE ... ADD COLUMN` unsettles reads on that table for minutes afterwards.
Migrate the test copy first, and do the live one at a quiet hour.
- **A migration that did nothing looks exactly like one that worked.** `CREATE TABLE IF NOT
EXISTS` succeeds whether or not it creates anything, so the applied/skipped counts say
nothing about whether the bucket now matches the code. A real case: `periods` predated
three columns the reopen path needs; migrate reported success, the columns were never
added, and the failure waited for the day an accountant reopened a period.
So the last thing this does is read the bucket back and compare it to schema.sql. A drift is a
non-zero exit, not a warning nobody reads.
erp-migrate # plan, apply, verify ./schema.sql
erp-migrate --schema bricks/quote/schema.sql --brick quote=0.3.0
erp-migrate --plan # print the plan and change nothing
erp-migrate --platform # the platform's own tables too (erp_public_link …), first
After a clean verify it records, in `platform_meta`, which erp-platform version migrated the
bucket and — with `--brick` — which version of each brick. At tier 1 one bucket holds several
bricks, so each has its own row (`brick:<name>`). An older version never overwrites a newer
record: an image rolled back runs against a schema a newer one left, and the record should keep
saying what the schema actually is.
"""
import os
import pathlib
import re
import sys
try:
from dotenv import load_dotenv
load_dotenv() # the credential lives in .env locally; deployed it is real env
except ImportError:
pass
from sqlalchemy import text
from erp_platform import db
from erp_platform import __version__ as PLATFORM_VERSION
def _arg(name: str) -> list:
"""Every value given for `--name VALUE`."""
out, argv = [], sys.argv
for i, a in enumerate(argv):
if a == name and i + 1 < len(argv):
out.append(argv[i + 1])
return out
def _version_tuple(v: str) -> tuple:
return tuple(int(x) if x.isdigit() else 0 for x in re.split(r"[.+-]", v)[:3])
def record_versions(bricks: dict) -> None:
"""Write platform_version and brick:<name> rows — never replacing a newer one with an older."""
from datetime import datetime, timezone
wanted = {"platform_version": PLATFORM_VERSION} | {f"brick:{n}": v for n, v in bricks.items()}
now, by = datetime.now(timezone.utc), os.environ.get("USER", "unknown")
with db.engine.begin() as conn:
conn.execute(text(STAMP_DDL))
with db.engine.begin() as conn:
stored = dict(conn.exec_driver_sql("SELECT key, value FROM platform_meta").fetchall())
for key, version in wanted.items():
if stored.get(key) and _version_tuple(stored[key]) > _version_tuple(version):
print(f"kept: {key} = {stored[key]} (this run is {version}, older — a rollback)")
continue
conn.execute(text(
"INSERT INTO platform_meta (key, value, set_at, set_by) VALUES (:k, :v, :t, :b) "
"ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, set_at = EXCLUDED.set_at, "
"set_by = EXCLUDED.set_by"), {"k": key, "v": version, "t": now, "b": by})
print(f"recorded: {key} = {version}")
def statements(sql: str):
"""Split on `;` at end of line — enough for DDL, and it keeps this file dependency-free."""
buf = []
for line in sql.splitlines():
if line.strip().startswith("--") or not line.strip():
continue
buf.append(line)
if line.rstrip().endswith(";"):
yield "\n".join(buf).rstrip().rstrip(";")
buf = []
def classify(stmt: str) -> str:
"""SAFE creates something new. DISRUPTIVE stalls reads on a live table. REFUSED changes a
shape an already-deployed app understands, which no rollback of the image can undo:
additive migrations are what make `ignite version rollback` an actual rollback."""
head = stmt.lstrip().upper()
if head.startswith("CREATE"):
return "SAFE"
if head.startswith("ALTER") and " ADD COLUMN" in head:
return "DISRUPTIVE"
return "REFUSED"
def expected_columns(sql: str) -> dict:
"""table -> [column], as schema.sql declares it, including the explicit ALTERs."""
out: dict = {}
for m in re.finditer(r'CREATE TABLE(?: IF NOT EXISTS)? ([a-z0-9_]+)\s*\(([\s\S]*?)\n\);', sql, re.I):
cols = []
for line in m.group(2).split("\n"):
# A comment — trailing, or a whole line of its own — is not a column. A comment-only line was
# once read as a column named "--" and reported as drift on a bucket that matched exactly.
line = re.sub(r"--.*$", "", line).strip().rstrip(",")
if not line or re.match(r'^(PRIMARY KEY|UNIQUE|FOREIGN KEY|CONSTRAINT|CHECK)\b', line, re.I):
continue
cols.append(line.split()[0])
out[m.group(1)] = cols
for m in re.finditer(r'ALTER TABLE ([a-z0-9_]+) ADD COLUMN ([a-z0-9_]+)', sql, re.I):
out.setdefault(m.group(1), []).append(m.group(2))
return out
def verify(sql: str) -> int:
# 0.9.11 note: ONE information_schema.columns query per schema file (the platform's + each brick's) — never a
# per-table inspector loop. On DataK3 a catalog query costs O(tables) today (dodilio/tables#494), so a composition
# of N bricks pays it N+1 times; it is the drift check that keeps a mismatched bucket from being recorded.
"""Read the bucket back. What the code expects must be what the bucket has."""
actual: dict = {}
with db.engine.connect() as conn:
rows = conn.execute(text(
"SELECT table_name, column_name FROM information_schema.columns")).all()
for table, column in rows:
actual.setdefault(table, set()).add(column)
drift = []
for table, cols in expected_columns(sql).items():
if table not in actual:
drift.append(f"missing table {table}")
continue
gap = [c for c in cols if c not in actual[table]]
if gap:
drift.append(f"missing column {table}: {', '.join(gap)}")
if drift:
print(f"\nDRIFT — the bucket does not match schema.sql ({len(drift)}):", file=sys.stderr)
for d in drift:
print(f" {d}", file=sys.stderr)
print("\nAdd the column to the CREATE *and* as an explicit ALTER below it.", file=sys.stderr)
return 1
print(f"verified: {db.BUCKET} matches schema.sql")
return 0
STAMP_DDL = """CREATE TABLE IF NOT EXISTS platform_meta (
key VARCHAR NOT NULL,
value VARCHAR,
set_at TIMESTAMP,
set_by VARCHAR,
PRIMARY KEY (key)
);"""
def stamp(environment: str) -> int:
"""Record what this bucket IS, so an app can refuse to use the wrong one.
A credential is not enough to tell dev from live. A k3 role CAN now be scoped to one bucket
(`grant-role … --on resource=<bucket drn>`; verified 2026-10-16 — the pg wire refuses the
others), and every deployed app should be — but a person's own login still reaches every
bucket in the org. An env var cannot tell either; a missing BUCKET is exactly how a test
suite once pointed at a live bucket. The only claim worth trusting is the one
the bucket makes about itself, so it is written here, once, deliberately.
"""
from datetime import datetime, timezone
with db.engine.begin() as conn:
conn.execute(text(STAMP_DDL))
with db.engine.begin() as conn:
existing = conn.exec_driver_sql(
"SELECT value FROM platform_meta WHERE key = 'environment'").fetchone()
if existing and existing[0] and existing[0] != environment:
print(f"refusing: {db.BUCKET} is already stamped {existing[0]!r}. Re-stamping a "
f"bucket is how a live one comes to be labelled 'dev' — if it is genuinely "
f"wrong, change it by hand and write down why.", file=sys.stderr)
return 1
conn.execute(text(
"INSERT INTO platform_meta (key, value, set_at, set_by) "
"VALUES ('environment', :v, :t, :b) "
"ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, set_at = EXCLUDED.set_at, "
"set_by = EXCLUDED.set_by"),
{"v": environment, "t": datetime.now(timezone.utc),
"b": os.environ.get("USER", "unknown")})
print(f"stamped: {db.BUCKET} is {environment!r}")
return 0
def main() -> int:
if "--stamp" in sys.argv:
i = sys.argv.index("--stamp")
if i + 1 >= len(sys.argv):
print("usage: migrate.py --stamp <dev|test|prod|demo>", file=sys.stderr)
return 1
return stamp(sys.argv[i + 1].strip().lower())
schema = pathlib.Path((_arg("--schema") or ["schema.sql"])[-1])
if not schema.is_file():
print(f"no schema at {schema.resolve()} — run from the brick's directory or pass "
"--schema <path>", file=sys.stderr)
return 1
bricks = {}
for spec in _arg("--brick"):
name, _, version = spec.partition("=")
if not (name and version):
print(f"--brick takes name=version, not {spec!r}", file=sys.stderr)
return 1
bricks[name] = version
if "--platform" in sys.argv:
code = migrate_platform(plan_only="--plan" in sys.argv)
if code:
return code
return migrate(schema, bricks, plan_only="--plan" in sys.argv)
def platform_schema() -> pathlib.Path:
"""erp-platform's own tables (platform.sql, shipped in the package): the public-link door's."""
import importlib.resources
return pathlib.Path(str(importlib.resources.files("erp_platform").joinpath("platform.sql")))
def migrate_platform(*, plan_only: bool = False) -> int:
"""Plan, apply and verify the platform's tables. A composition runs it before its bricks; a brick served
on its own runs `erp-migrate --platform` when it issues public links."""
print("── erp-platform's own tables ──")
return migrate(platform_schema(), {}, plan_only=plan_only)
def migrate(schema, bricks: dict, *, plan_only: bool = False) -> int:
"""Plan, apply and verify one schema file, then record the versions. What `erp-migrate` runs, and
what a composition runs once per brick (erp_platform.compose)."""
sql = pathlib.Path(schema).read_text()
plan = [(classify(s), s) for s in statements(sql)]
counts = {k: sum(1 for c, _ in plan if c == k) for k in ("SAFE", "DISRUPTIVE", "REFUSED")}
print(f"plan for {db.BUCKET}: {counts['SAFE']} safe, {counts['DISRUPTIVE']} disruptive, "
f"{counts['REFUSED']} refused")
for kind, stmt in plan:
if kind == "DISRUPTIVE":
print(f" DISRUPTIVE {stmt.splitlines()[0][:70]} (reads on this table stall for minutes)")
if kind == "REFUSED":
print(f" REFUSED {stmt.splitlines()[0][:70]}", file=sys.stderr)
if counts["REFUSED"]:
print("\nRefused: a DROP, RENAME or type change is not additive, so an app rolled back to\n"
"the previous image would meet a schema it does not understand. Do it as a separate\n"
"release after the new code has been live and the rollback window has closed.",
file=sys.stderr)
return 1
if plan_only:
return 0
applied = skipped = 0
for _, stmt in plan:
try:
with db.engine.begin() as conn:
conn.execute(text(stmt))
applied += 1
except Exception as exc: # noqa: BLE001
msg = str(exc).lower()
# already-there is the normal case on a re-run, not a failure
if "already exists" in msg or "duplicate column" in msg:
skipped += 1
continue
print(f"FAILED: {stmt.splitlines()[0][:80]}\n {exc}", file=sys.stderr)
return 1
print(f"applied {applied} statement(s), {skipped} already present, bucket {db.BUCKET}")
drift = verify(sql)
if drift:
return drift # a bucket that does not match the code is not recorded as migrated
record_versions(bricks)
return 0
def seed(member) -> int:
"""0.9.11: the rows a NEW bucket starts with — the brick's own `Repository.seed()` (payment terms, units, product
types, order types, payment settings…), after its schema, the way `migrate --local` runs it. Every brick's seed
inserts only what is missing, so a re-run adds nothing. Before 0.9.11 a bucket was migrated without them and the
small-ERP deploy seeded by hand (erp-compositions smallerp-demo scripts/live/seed_bricks.py)."""
import importlib
import time
from erp_platform.scenarios.inprocess import root_repository
try:
tables = importlib.import_module(f"{member.package}.infrastructure.tables")
except ImportError:
print(f"seed: {member.name} has no {member.package}.infrastructure.tables — nothing to seed")
return 0
try:
repo = root_repository(member.name, tables)
except RuntimeError as e:
print(f"seed: {e}", file=sys.stderr)
return 1
if not callable(getattr(repo, "seed", None)):
print(f"seed: {member.name} starts with no rows (its repository has no seed())")
return 0
started = time.time()
try:
with db.engine.begin() as conn:
repo(conn).seed()
except Exception as exc: # noqa: BLE001
print(f"seed FAILED: {member.name}: {type(exc).__name__}: {str(exc)[:300]}", file=sys.stderr)
return 1
print(f"seeded {member.name}'s starting rows in {time.time() - started:.1f}s (re-run safe: only what is missing)")
return 0
if __name__ == "__main__":
raise SystemExit(main())