"""erp-compose seed — replay configuration calls through the composition's routes, re-runnable.
erp-compose seed --local [DIR] compose.yaml config/starter.calls.yaml the dev tier (SQLite, what migrate --local made)
BUCKET=<bucket> erp-compose seed compose.yaml config/starter.calls.yaml a DataK3 bucket, in process
0.9.10. The 2026-10-25 timed lab run (a small UK wholesaler, 8 bricks) spent most of its 74 turns writing an MCP
client, a shell wrapper and setup scripts by hand to create a chart, accounts, the account maps, tax codes, a
warehouse, a bank account… one call at a time, restarting when one was refused. This is that, as one command over
the file `erp-compose process` already replays: a YAML list of `{use_case: <name>, args: {...}}` (or a mapping with
`calls:`). A call may save its answer (`as: acct_1200`) for a later argument to read (`"$acct_1200.account_id"`);
`$today` / `$tomorrow` / `$year` are the business day (erp_platform.process_export.resolve).
RE-RUNNABLE. Each command is sent with an Idempotency-Key made from its use case and arguments, so a command that
makes something new on every call answers its first answer again (`Idempotent-Replayed`) instead of making a
second; a keyed upsert (keep_*, put_*) answers 200 with the row as it is. Either way: "already there". A 409 is a
row that already exists in a form the brick will not take twice — skipped and said so, with the brick's reason.
Anything else stops the seed at that call, naming it, its status and the brick's reason: the calls before it are
in, and a re-run after the fix skips them.
DAY ONE. Bricks take many rows only from a date after today (a correction is a new dated row); a starter dates those
"$tomorrow". `seed --local` records the business day it ran on in the dev tier's marker, and `serve --local` with
no --clock then starts the business clock on the next day, so a sale can run straight after the seed (and says so).
WHO IT ACTS AS: the dev owner — every permission the composed bricks declare, machine ones included
(erp_platform.local.owner_identity) — presented the way the gateway would. Through the routes, never an INSERT
into a brick's table: every rule a person would meet is met.
WHERE IT RUNS. Both tiers boot the composition IN PROCESS and call it through ASGI, the way `migrate` reaches the
database from the operator's machine: `--local` on the SQLite files of `migrate --local` (run it while `serve
--local` is stopped or running — SQLite takes both); without it, on the bucket named by BUCKET with the app's own
environment (the pg wire, a service-account token — what `erp-compose serve` reads). The deployed app's URL is not
used: behind the gateway a caller is a pool user, and no pool user holds machine permissions.
"""
import hashlib
import json
import os
import pathlib
import sys
from contextlib import ExitStack
USAGE = ("usage: erp-compose seed [--local [DIR]] <compose.yaml> <CALLS.yaml>\n"
" Replay configuration calls ({use_case, args} list) through the routes as the dev owner. Re-runnable: a\n"
" call already applied answers 200 (replayed) or 409 and is skipped; the first other refusal stops it.\n"
" --local [DIR]: the SQLite files `migrate --local` made. Without it: the bucket named by BUCKET, in process.")
def key_for(use_case: str, args: dict) -> str:
body = json.dumps(args, sort_keys=True, separators=(",", ":"), default=str)
return "seed-" + hashlib.sha256(f"{use_case}\n{body}".encode()).hexdigest()[:40]
def replay(app, comp, rows: list, me: dict, say=print, source: str = "calls") -> dict:
"""Send each row; returns {applied, upserted, already, skipped, total}. Raises RuntimeError at the first real
refusal. Says one line per run of the same use case with the same outcome (25 accounts are one line)."""
from erp_platform.process_export import reason, resolve, send, use_cases
ucs = use_cases(comp)
saved, counts, total = {}, {"applied": 0, "upserted": 0, "already": 0, "skipped": 0}, len(rows)
run = [] # [first n, last n, use case, outcome] not yet said
def flush():
if run:
a, b, name, what = run
say(f" {a if a == b else f'{a}-{b}':>7} {name:<30} " + (what if a == b else f"x{b - a + 1} {what}"))
run.clear()
def note(n, name, what, alone=False):
if run and run[2] == name and run[3] == what and not alone:
run[1] = n
return
flush()
run.extend([n, n, name, what])
if alone:
flush()
for n, row in enumerate(rows, 1):
name = row["use_case"]
where = f"{source} call {n}/{total} ({name})"
try:
args = resolve(dict(row.get("args") or {}), saved)
except KeyError as e:
flush()
raise RuntimeError(f"{where}: {e.args[0]}") from None
try:
r = send(app, ucs, me, name, args, key=key_for(name, args))
except LookupError as e:
flush()
raise RuntimeError(f"{where}: {e}") from None
if 200 <= r.status_code < 300:
replayed, upsert = r.headers.get("Idempotent-Replayed") == "true", ucs[name][1].idempotent
counts["already" if replayed else "upserted" if upsert else "applied"] += 1
note(n, name, "already there (the first answer, replayed)" if replayed else
"ok (keyed: a re-run writes the same row)" if upsert else "ok")
if row.get("as"):
try:
saved[str(row["as"])] = r.json()
except ValueError:
saved[str(row["as"])] = None
elif r.status_code == 409:
counts["skipped"] += 1
note(n, name, f"skipped (409, taken as already there — check it is): {reason(r)}", alone=True)
else:
flush()
raise RuntimeError(f"{where}: refused {r.status_code}: {reason(r)}")
flush()
return {**counts, "total": total}
def run(compose_path: str, calls_path: str, *, local: bool, local_dir=None, say=print) -> int:
from erp_platform import compose
from erp_platform import local as local_tier
from erp_platform.process_export import _calls
root = pathlib.Path(compose_path).resolve().parent
compose.brick_paths(root)
if local:
try:
local_tier.guard()
except local_tier.LocalRefused as e:
print(e, file=sys.stderr)
return 1
elif not os.environ.get("BUCKET", "").strip():
print("erp-compose seed: name the bucket (BUCKET=<bucket>, with the app's pg-wire env and token), "
"or seed the dev tier with --local", file=sys.stderr)
return 2
try:
comp = compose.load(compose_path)
except compose.CompositionRefused as e:
print(e, file=sys.stderr)
return 1
rows = _calls(pathlib.Path(calls_path))
me = local_tier.owner_identity(comp)
with ExitStack() as st:
if local:
data = local_tier.data_dir(root, local_dir)
if not (data / local_tier.MARKER).exists():
print(f"{data}/ is not migrated: run `erp-compose migrate --local` first", file=sys.stderr)
return 1
say(f"LOCAL DEV TIER — erp-compose seed --local: {len(rows)} call(s) from {calls_path} into {data}/, "
f"as the dev owner ({len(me['permissions'])} permissions)")
app = local_tier._stage(comp, st, data, create=False).app
else:
me = {**me, "sub": "erp-compose.seed", "email": "seed@erp-compose.local"}
say(f"erp-compose seed: {len(rows)} call(s) from {calls_path} into bucket {os.environ['BUCKET']}, "
"in process with this shell's app environment")
app = compose.app(comp)
try:
res = replay(app, comp, rows, me, say, source=str(calls_path))
except RuntimeError as e:
print(f"seed stopped — {e}\n The calls before it are in; fix this one (or the data it needs) and re-run: "
"what is already there is skipped.", file=sys.stderr)
return 1
if local:
local_tier.mark_seeded(data, str(calls_path))
say(f"seeded {res['total']} call(s): {res['applied']} applied, {res['upserted']} keyed upsert(s), "
f"{res['already'] + res['skipped']} already there ({res['already']} replayed, {res['skipped']} answered 409)")
return 0
def main(argv) -> int:
args = list(argv)
if not args or any(a in ("-h", "--help") for a in args):
print(USAGE)
return 0 if args else 2
local, local_dir = False, None
for i, a in enumerate(args):
if a == "--local" or a.startswith("--local="):
local, n = True, 1
if "=" in a:
local_dir = a.split("=", 1)[1] or None
elif i + 1 < len(args) and not args[i + 1].startswith("-") and not args[i + 1].endswith((".yaml", ".yml", ".json")):
local_dir, n = args[i + 1], 2
del args[i:i + n]
break
if len(args) != 2 or any(a.startswith("-") for a in args):
print(USAGE, file=sys.stderr)
return 2
return run(args[0], args[1], local=local, local_dir=local_dir)