"""scaffold — a module spec in, a brick skeleton out, in the exact shape of templates/brick-example.
erp-compose scaffold modules/contracts/module.yaml --out ../brick-contracts [--entity contract …]
[--dist erp-brick-contracts] [--force]
DETERMINISTIC: no model is called, the same spec gives the same bytes, and nothing in the output depends on the
clock or the machine. What an agent then writes is only what the spec cannot decide: the business rules
(`TODO(rule)`), row visibility where the spec says it in prose (`TODO(visibility)`), permissions the spec does
not name (`TODO(permission)`), the scenario steps (`TODO(scenario)`) and AI-step modes (`TODO(ai)`).
`grep -rn "TODO(" src tests` is the agent's work list; `pytest` shows one xfail per story still open.
What the spec decides, the scaffold writes:
entities → one domain module per aggregate root (an entity whose natural key extends another's is a
part of that one's aggregate), typed dataclasses, value objects for enumerated states and
attributes, `check_<story>` invariant stubs raising RuleBroken, one event per transition;
the tables (prefixed, a PRIMARY KEY each, money NUMERIC(18,2), the `<root>_versions`
claim table), one repository per aggregate, schema.sql and migrations/00001_init from the
SAME column model;
workflows → one use case per transition (route + MCP tool), get/list queries per root, `keep_<root>`
for a root no transition creates, `if_match` on every command, process.yaml so the brick's
BPMN is the spec's (`erp-compose bpmn --from module.yaml`);
permissions → the strings on each use case, resolved from the actors that may act and the catalog;
config_rows → the spec's settings table, ONE typed row per setting, seeded, with list/set use cases;
a configuration entity is a keyed table edited under the settings permission, an
append-only history entity is appended on every save (0.9.1: neither is an aggregate);
stories → tests/test_rules.py, one strict xfail per story; tests.scenarios → a catalogue skeleton
that runs in process with the permission matrix; tests/test_probes.py → the generated
probes (roles × commands, owners, state guards, terminal parents, separation of duties,
seams, replay, mass assignment), xfail(strict) where the rule is the brick's;
ai_steps → seams on the ai.* capabilities (named, or inferred from the step), the mode from the spec,
`propose()` and the audit fields, the manual fallback (erp-roadmap #53).
It never overwrites a file that differs without --force. `scaffold --update` (re-scaffolding a brick that has been
filled in) is out of scope: the skeleton is a starting point, owned by the brick once written.
"""
import argparse
import json
import keyword
import pathlib
import re
import sys
import textwrap
from dataclasses import dataclass, field
from typing import Optional
import yaml
from erp_platform import bpmn
PLATFORM_PIN = "erp-platform>=0.9.2,<0.10"
# modules/shared.yaml (erp-method), copied: the scaffold must run without the method checked out.
BIGINT_KEYS = {"bp_id", "account_id", "journal_id", "person_id"}
MONEY = re.compile(r"amount|total|price|balance|debit|credit|subtotal|cost")
NOT_MONEY_SUFFIXES = ("_id", "_no", "_code", "_key", "_book", "_pct", "_centre", "_center", "_mode", "_ref", "_by",
"_on", "_at")
ALSO_NOT_MONEY = {"normal_balance"}
RESERVED = {"caller", "by", "outcome", "repo", "agg", "found", "current", "it", "events", "created_at", "updated_at",
"self", "cls", "attrs"}
IDENT = re.compile(r"[a-z_][a-z0-9_]*")
HISTORY = re.compile(r"append[- ]only|\bhistory\b|\baudit (?:trail|log)\b")
MODES = ("off", "suggest", "auto-with-review", "auto")
TODO_PERM = "TODO(permission):" # a permission the spec's catalog does not name: held by no one
SETTING_STOP = {"a", "an", "the", "and", "or", "of", "for", "per", "whether", "which", "each", "what", "make",
"makes", "made", "offered", "exists", "exist", "is", "are", "be", "to", "in", "on", "one", "its",
"it", "by", "with", "from", "how", "who", "when", "that", "this", "their", "there", "may", "can"}
VALUE = re.compile(r"[A-Za-z0-9_\-]+")
KINDS = { # kind: (python annotation, sqlalchemy type, SQL type, python default, sample JSON value)
"str": ("str", "String", "VARCHAR", '""', None),
"int": ("int", "Integer", "INTEGER", "0", 1),
"bigint": ("int", "BigInteger", "BIGINT", "0", 1001),
"bool": ("bool", "Boolean", "BOOLEAN", "False", True),
"decimal": ("Decimal", "Numeric(18, 2)", "NUMERIC(18,2)", 'Decimal("0.00")', "1.00"),
"rate": ("Decimal", "Numeric(19, 4)", "NUMERIC(19,4)", 'Decimal("0.0000")', "1.0000"),
"pct": ("Decimal", "Numeric(9, 4)", "NUMERIC(9,4)", 'Decimal("0.0000")', "1"),
"date": ("date", "Date", "DATE", "None", "2026-01-01"),
"datetime": ("datetime", "TIMESTAMP", "TIMESTAMP", "None", "2026-01-01T09:00:00Z"),
}
class ScaffoldError(Exception):
"""The spec cannot be scaffolded as it stands; the message says which part."""
# ── naming ─────────────────────────────────────────────────────────────────────────────────
def camel(s) -> str:
return "".join(p[:1].upper() + p[1:].lower() for p in re.split(r"[^A-Za-z0-9]+", str(s)) if p)
def slug(s) -> str:
return re.sub(r"[^a-z0-9]+", "_", str(s).lower()).strip("_")
def plural(s: str) -> str:
if s.endswith("y") and s[-2:-1] not in ("a", "e", "i", "o", "u"):
return s[:-1] + "ies"
if s.endswith(("s", "x", "sh", "ch")):
return s + "es"
return s + "s"
def human(s) -> str:
return str(s).replace("_", " ").replace("-", " ").lower()
def a_(word: str) -> str:
return ("an " if str(word)[:1].lower() in "aeiou" else "a ") + str(word)
def member(v: str) -> str:
m = re.sub(r"[^A-Za-z0-9]+", "_", str(v)).strip("_").upper() or "NONE"
return "S_" + m if m[0].isdigit() else m
def lit(s) -> str:
"""A Python string literal (JSON's escapes are Python's)."""
return json.dumps(str(s), ensure_ascii=False)
def clean(text) -> str:
return " ".join(str(text or "").split())
def docstring(text: str, indent: int, width: int = 112) -> str:
"""A triple-quoted docstring, wrapped, safe for any spec text."""
t = clean(text).replace("\\", "\\\\").replace('"""', "'''") or "TODO(doc)."
if t.endswith('"'):
t += "."
pad = " " * indent
lines = textwrap.wrap(t, width - indent - 3) or [t]
if len(lines) == 1:
return f'{pad}"""{lines[0]}"""\n'
return f'{pad}"""' + f"\n{pad}".join(lines) + f'\n{pad}"""\n'
# ── the model ──────────────────────────────────────────────────────────────────────────────
@dataclass
class Field:
name: str
kind: str = "str"
enum: tuple = ()
vo: str = ""
dflt: str = "" # a state field's initial state
@property
def py(self) -> str:
return KINDS[self.kind][0]
@property
def default(self) -> str:
if self.vo and self.enum:
return f"{self.vo}.{member(self.dflt or self.enum[0])}.value"
return KINDS[self.kind][3]
@property
def nullable(self) -> bool:
return self.kind in ("date", "datetime")
@dataclass
class Ent:
id: str
meaning: str
key: list
attrs: list
table: str
parent: Optional["Ent"] = None
children: list = field(default_factory=list)
state: Optional[Field] = None
states: list = field(default_factory=list)
initial: str = ""
synthetic: bool = False
steps: list = field(default_factory=list)
stories: list = field(default_factory=list)
opt: str = "version"
role: str = "aggregate" # 0.9.1: aggregate | config (keep_state) | history (keep_event)
terminal: list = field(default_factory=list)
@property
def cls(self) -> str:
return camel(self.id)
@property
def root(self) -> "Ent":
return self.parent.root if self.parent else self
@property
def descendants(self) -> list:
out = []
for c in self.children:
out += [c, *c.descendants]
return out
@property
def fields(self) -> list:
return [*self.key, *self.attrs]
@property
def list_attr(self) -> str:
return plural(self.id)
@property
def extra_key(self) -> list:
"""The key fields beyond the root's: what finds this part inside its aggregate."""
n = len(self.root.key)
return self.key[n:]
def has(self, name: str) -> bool:
return any(f.name == name for f in self.fields)
@dataclass
class Step:
key: str
ent: Ent
frm: list
to: list
by: list
gate: str
requires: str
requires_text: str
effect: str
task: str
event: str = ""
permission: str = ""
permission_todo: bool = False
@property
def creates(self) -> bool:
return "none" in self.frm
@property
def decides(self) -> bool:
return len(self.to) > 1
@property
def loop(self) -> bool:
return len(self.to) == 1 and self.to[0] in self.frm
@dataclass
class Setting:
key: str
kind: str # str | int | bool | date | mode
changes: str
choices: tuple = () # kind mode: the modes the step allows
default: object = None
@dataclass
class Story:
id: str
kind: str
story: str
given: str
when: str
then: str
root: Optional[Ent] = None
@property
def check(self) -> str:
return "check_" + slug(self.id)
@dataclass
class Model:
spec: dict
id: str
name: str
summary: str
ns: str
const: str
package: str
dist: str
prefix: str
roots: list
steps: list
stories: list
catalog: list
actors: list
config_rows: list
scenarios: list
ai_steps: list
row_rule: str
depends_on_bp: bool
has_process: bool
read_perm: dict = field(default_factory=dict)
write_perm: dict = field(default_factory=dict)
todo_perms: set = field(default_factory=set)
skipped: list = field(default_factory=list)
settings_table: str = ""
settings: list = field(default_factory=list) # [Setting]: one row per setting, typed
configs: list = field(default_factory=list) # [Ent]: configuration tables (keep_state), edited as settings
histories: list = field(default_factory=list) # [Ent]: append-only history (keep_event), appended on save
seams: list = field(default_factory=list) # [(with, [step keys])]: the outbound seams and their steps
def kind_of(name: str) -> str:
if name in BIGINT_KEYS:
return "bigint"
if name == "rate_amount":
return "rate"
if name in ("n", "seq", "version", "revision", "position", "quantity") or \
name.endswith(("_no", "_days", "_count", "_level", "_seq", "_months", "_hours")):
return "int"
if name.endswith(("_on", "_date")) or name in ("valid_from", "valid_to", "date"):
return "date"
if name.endswith("_at"):
return "datetime"
if name.startswith(("is_", "has_")) or name.endswith(("_allowed", "_required", "_enabled")) or \
name in ("required", "active", "enabled"):
return "bool"
if name.endswith("_pct") or name == "tax_rate":
return "pct"
if MONEY.search(name) and not name.endswith(NOT_MONEY_SUFFIXES) and name not in ALSO_NOT_MONEY:
return "decimal"
return "str"
def safe_name(name: str) -> str:
return name + "_" if keyword.iskeyword(name) else name
def parse_key(raw, entity_id: str) -> list:
"""`a|b`, `x = derivation`, `a|b (note)` or a list → field names; anything else → `<entity>_id`."""
if isinstance(raw, list):
parts = [str(p) for p in raw]
else:
text = re.sub(r"\([^)]*\)", "", str(raw or ""))
text = text.split("=", 1)[0].strip()
parts = [p.strip() for p in text.split("|")]
names = [safe_name(p.lower()) for p in parts]
if not names or not all(IDENT.fullmatch(n) for n in names):
return [f"{entity_id}_id"]
return names
def parse_attr(raw) -> Optional[Field]:
text = clean(raw)
m = re.match(r"^([A-Za-z_][A-Za-z0-9_]*)\s*(?:\(([^)]*)\))?", text)
if not m:
return None
name = safe_name(m.group(1).lower())
enum = ()
if m.group(2):
vals = [v.strip() for v in m.group(2).split("|")]
if len(vals) >= 2 and all(VALUE.fullmatch(v) for v in vals):
enum = tuple(vals)
return Field(name, "str" if enum else kind_of(name), enum)
def _parent_of(e: Ent, ents: list) -> Optional[Ent]:
names = [f.name for f in e.key]
best = None
for o in ents:
on = [f.name for f in o.key]
if o is e or len(on) >= len(names) or names[:len(on)] != on:
continue
if best is None or len(on) > len(best.key):
best = o
elif len(on) == len(best.key):
# equal depth: the parent whose name ends the child's (person_sensitive → sensitive), else declared first
if e.id.endswith(o.id) and not e.id.endswith(best.id):
best = o
return best
# ── 0.9.1: AI-step modes, settings, seams ──────────────────────────────────────────────────
def _spec_line(text: Optional[str], step: str, field_: str) -> str:
"""' (module.yaml line n)': the `field_` line of ai step `step`, found in the source text."""
if not text:
return ""
lines = text.splitlines()
top = next((i for i, ln in enumerate(lines) if ln.startswith("ai_steps:")), 0)
start = next((i for i, ln in enumerate(lines) if i >= top and re.search(rf"\b(?:key|step):\s*['\"]?{re.escape(step)}\b", ln)),
None)
if start is None:
return ""
for i in range(start, min(len(lines), start + 40)):
if re.search(rf"\b{field_}\s*:", lines[i]):
return f" (module.yaml line {i + 1})"
return f" (module.yaml line {start + 1})"
def _mode(v, step: str, field_: str, text: Optional[str]) -> str:
"""A mode as the spec meant it. YAML 1.1 reads an unquoted `off` as false: that is normalised back to "off".
`on`/`yes`/`true` (read as true) is not a mode, and neither is any other word: refused, naming the line."""
if v is False:
return "off"
m = str(v).strip().lower().replace("_", "-") if not isinstance(v, bool) else None
if m not in MODES:
hint = (" — unquoted on/yes/true is YAML for true; write the mode, quoted (\"off\")" if v is True else "")
raise ScaffoldError(f"ai_steps {step}: {field_} {v!r} is not a mode ({' | '.join(MODES)})"
f"{_spec_line(text, step, field_)}{hint}")
return m
def _ai_steps(spec: dict, text: Optional[str]) -> list:
out = []
for a in spec.get("ai_steps") or []:
if not isinstance(a, dict):
continue
a = dict(a)
step = slug(a.get("key") or a.get("step") or "step")
if "modes_allowed" in a and a["modes_allowed"] is not None:
raw = a["modes_allowed"] if isinstance(a["modes_allowed"], list) else [a["modes_allowed"]]
a["modes_allowed"] = [_mode(x, step, "modes_allowed", text) for x in raw]
if a.get("mode") is not None:
a["mode"] = _mode(a["mode"], step, "mode", text)
out.append(a)
return out
def _setting_kind(text: str, declared=None) -> str:
d = str(declared or "").lower()
if d:
return {"integer": "int", "boolean": "bool", "string": "str", "text": "str"}.get(d, d) \
if d in ("int", "integer", "bool", "boolean", "date", "str", "string", "text") else "str"
t = text.lower()
if re.search(r"\bwhether\b|\bon/off\b|\bon or off\b|\benabled?\b|\ballowed\b", t):
return "bool"
if re.search(r"\b(?:working )?(?:days?|hours?|weeks?|months?)\b|\bnumber of\b|\bhow many\b|\bcount\b|"
r"\blimit\b|\bthreshold\b", t):
return "int"
if re.search(r"\bdates?\b|\bholidays?\b", t):
return "date"
return "str"
def _clause_key(clause: str) -> str:
words = [w for w in re.findall(r"[a-z0-9]+", re.sub(r"\([^)]*\)", "", clause.lower())) if w not in SETTING_STOP]
return "_".join(words[:3]) or "setting"
def _settings(m: "Model", spec: dict) -> list:
"""One row per SETTING (0.9.1), typed. A config_rows entry is, in order: a `settings:` list (key, type,
changes — exact), one setting by `key:`, the settings table's prose (one setting per `;` clause, and per
"…, and whether …"; "each AI step's mode" is one `ai.<step>.mode` row per ai step), a configuration table of
its own (pj_state: its rows, not a setting), else the legacy one-table-one-setting."""
own = {e.table for e in m.configs} | {e.table for e in m.histories} | \
{e.table for r in m.roots for e in (r, *r.descendants)}
out = []
for c in m.config_rows:
t = slug(c.get("table") or "")
changes = clean(c.get("changes"))
if isinstance(c.get("settings"), list):
for x in c["settings"]:
if isinstance(x, dict) and x.get("key"):
out.append(Setting(str(x["key"]), _setting_kind(clean(x.get("changes")), x.get("type")),
clean(x.get("changes") or changes), default=x.get("default")))
elif c.get("key"):
out.append(Setting(str(c["key"]), _setting_kind(changes, c.get("type")), changes))
elif t and t in own:
continue
elif t == m.settings_table:
for part in [p.strip() for p in changes.split(";") if p.strip()]:
for i, clause in enumerate(re.split(r",?\s+and (?=whether\b)", part)):
if re.search(r"\bai\b.*\bmodes?\b", clause.lower()):
out += [Setting(f"ai.{slug(a.get('key') or a.get('step'))}.mode", "mode",
f"{clean(a.get('benefit')) or 'the AI step'}: " + " or ".join(
a.get("modes_allowed") or ["off", "suggest"]),
tuple(a.get("modes_allowed") or ["off", "suggest"]))
for a in m.ai_steps if a.get("key") or a.get("step")]
continue
out.append(Setting(_clause_key(clause), _setting_kind(clause), clause))
else:
out.append(Setting(_setting_key(m, c), _setting_kind(changes, c.get("type")), changes))
seen = set()
for st in out:
k, n = st.key, 2
while st.key in seen:
st.key, n = f"{k}_{n}", n + 1
seen.add(st.key)
return out
def _seams(m: "Model", spec: dict) -> list:
"""[(brick it calls, [step keys])]: the OUTBOUND seams (owned elsewhere, or "this module calls …") and the
transitions that fire them — matched by the story's trigger words against the step keys. A starting point:
the probes assert a seam is called ONLY on these steps, so the brick corrects the list, not the probe."""
stories = {str(x.get("id")): x for x in spec.get("stories") or [] if isinstance(x, dict)}
out = []
for sm in spec.get("seams") or []:
if not isinstance(sm, dict) or not sm.get("with"):
continue
rule = clean(sm.get("rule")).lower()
if str(sm.get("owner") or "") == m.id and not re.search(r"this module calls|go(?:es)? back through", rule):
continue
st = stories.get(str(sm.get("story")), {})
words = set(re.findall(r"[a-z]+", (clean(st.get("trigger")) or clean(st.get("story"))).lower()))
steps = [x.key for x in m.steps if any(any(w.startswith(k) for w in words)
for k in x.key.split("_") if len(k) >= 4)
and (human(x.ent.id) in words or x.ent.id in words)]
out.append((slug(str(sm["with"])).replace("_", "-"), steps))
return out
def build(spec: dict, *, entities: Optional[list] = None, dist: Optional[str] = None,
text: Optional[str] = None) -> Model:
if not isinstance(spec, dict) or not spec.get("id"):
raise ScaffoldError("the spec has no `id`")
mid = str(spec["id"])
brick = spec.get("brick") or {}
perms = spec.get("permissions") or {}
ns = str(brick.get("short_name") or perms.get("namespace") or slug(mid))
package = str(brick.get("package") or "brick_" + slug(mid))
raw_ents = [e for e in spec.get("entities") or [] if isinstance(e, dict) and e.get("id")]
tables = [str(e.get("reference_table") or "") for e in raw_ents if e.get("reference_table")]
prefix = str(brick.get("prefix") or "")
if not prefix:
heads = [t.split("_", 1)[0] + "_" for t in tables if "_" in t]
prefix = max(heads, key=lambda h: (heads.count(h), -heads.index(h))) if heads else slug(ns) + "_"
# 0.9.1: the settings table is the spec's (its `settings` entity's reference_table: pj_setting), never invented
settings_ent = next((e for e in raw_ents if slug(e["id"]) in ("settings", "setting")), None)
settings_table = slug(settings_ent.get("reference_table") or "") if settings_ent else ""
if not settings_table:
named = [slug(c.get("table")) for c in spec.get("config_rows") or [] if isinstance(c, dict)
and re.fullmatch(r".*_settings?", slug(c.get("table") or ""))]
settings_table = named[0] if named else f"{prefix}settings"
ents, skipped = [], []
for e in raw_ents:
eid = slug(e["id"])
if e is settings_ent:
continue # the settings table is generated from config_rows
if re.search(r"[,\s]", str(e.get("reference_table") or "").strip()):
skipped.append(eid) # one entity naming several tables: a catalogue, by hand
continue
table = slug(e.get("reference_table") or prefix + plural(eid))
if table == settings_table or eid == "settings":
continue # the settings table is generated from config_rows
if not table.startswith(prefix):
table = prefix + table
key = [Field(n, kind_of(n)) for n in parse_key(e.get("natural_key"), eid)]
attrs, seen = [], {f.name for f in key}
for a in e.get("key_attributes") or []:
f = parse_attr(a)
if f is None or f.name in seen or f.name in RESERVED:
continue
seen.add(f.name)
attrs.append(f)
role = ("config" if "configuration" in str(e.get("master_data") or "").lower() else
"history" if HISTORY.search(clean(e.get("meaning")).lower()) else "aggregate")
ents.append(Ent(eid, clean(e.get("meaning")), key, attrs, table, role=role))
if not ents and not spec.get("workflows"):
raise ScaffoldError(f"{mid}: no entities and no workflows — nothing to scaffold")
# workflows bind to an entity (exact id, else the entity whose id ends `_<entity>`), else a synthetic root
wfs = [w for w in spec.get("workflows") or [] if isinstance(w, dict) and w.get("entity") and w.get("transitions")]
bound = {}
for w in wfs:
we = slug(w["entity"])
e = next((x for x in ents if x.id == we), None) or next((x for x in ents if x.id.endswith("_" + we)), None)
if e is not None and e.role != "aggregate":
e.role = "aggregate" # a lifecycle makes it an aggregate, whatever its words
if e is None:
e = Ent(we, f"the {human(we)} (no entity in the spec: TODO(rule) describe it)",
[Field(f"{we}_id", "str")], [], prefix + plural(we), synthetic=True)
ents.append(e)
bound[w["entity"]] = e
configs = [e for e in ents if e.role == "config"]
histories = [e for e in ents if e.role == "history"]
ents = [e for e in ents if e.role == "aggregate"]
for e in ents:
e.parent = _parent_of(e, ents)
if e.parent:
e.parent.children.append(e)
roots = [e for e in ents if e.parent is None]
if entities:
wanted = [slug(x) for x in entities]
unknown = [x for x in wanted if not any(e.id == x for e in ents)]
if unknown:
raise ScaffoldError(f"--entity {', '.join(unknown)}: not an entity of {mid} "
f"(there are: {', '.join(e.id for e in ents)})")
keep = {next(e for e in ents if e.id == x).root.id for x in wanted}
roots = [r for r in roots if r.id in keep]
configs = [e for e in configs if e.id in wanted]
histories = [e for e in histories if e.id in wanted]
selected = {id(x) for r in roots for x in (r, *r.descendants)}
# states and transitions
catalog = [str(p) for p in perms.get("catalog") or []]
actors = [a for a in spec.get("actors") or [] if isinstance(a, dict) and a.get("id")]
model = Model(spec, mid, str(spec.get("name") or mid), clean(spec.get("summary")), ns,
re.sub(r"[^A-Za-z0-9]", "_", ns).upper(), package, dist or f"brick-{mid}", prefix, roots, [], [],
catalog, actors, [c for c in spec.get("config_rows") or [] if isinstance(c, dict)],
list((spec.get("tests") or {}).get("scenarios") or []), list(spec.get("ai_steps") or []),
clean((perms.get("row_visibility") or {}).get("rule")),
any((d or {}).get("module") == "business-partner"
for d in (spec.get("facets") or {}).get("depends_on") or []),
bool(wfs))
for w in wfs:
e = bound[w["entity"]]
states = [str(s) for s in w["states"]] if isinstance(w.get("states"), list) else []
for t in w["transitions"]:
for s in bpmn._list(t.get("from")) + bpmn._list(t.get("to")):
if s != "none" and s not in states:
states.append(s)
e.states = states
sname = next((n for n in ("state", "status") if any(f.name == n for f in e.attrs)), "status")
e.attrs = [f for f in e.attrs if f.name != sname]
e.state = Field(sname, "str", tuple(states), f"{e.cls}State")
creates = [t for t in w["transitions"] if "none" in bpmn._list(t.get("from"))]
e.initial = (bpmn._list(creates[0].get("to")) or [""])[0] if creates else (states[0] if states else "")
e.state.dflt = e.initial
outs = {s for t in w["transitions"] for s in bpmn._list(t.get("from"))}
e.terminal = [str(x) for x in bpmn._list(w.get("terminal"))] or [s for s in states if s not in outs]
if id(e) not in selected:
continue
for t in w["transitions"]:
frm, to = bpmn._list(t.get("from")), bpmn._list(t.get("to"))
to = to or (frm if len(frm) == 1 else [])
key = str(t.get("key") or bpmn.derived_key(w["entity"], frm, to))
req = clean(t.get("requires"))
req_id = safe_name(req.lower()) if IDENT.fullmatch(req.lower() or "-") else ""
tr = bpmn.Transition(key=key, frm=frm, to=to, by=bpmn._list(t.get("by")), name=str(t.get("name") or ""))
st = Step(key, e, frm, to, bpmn._list(t.get("by")), clean(t.get("gate")), req_id, req,
clean(t.get("effect")), bpmn.task_name(tr, w["entity"]))
e.steps.append(st)
model.steps.append(st)
for e in ents:
for f in e.attrs:
if f.enum:
f.vo = e.cls + camel(f.name)
# optimistic version column: ALWAYS `version` (0.9.1). A spec that lists `version` on the root means the same
# column — it is reused, not shadowed by an invented `revision`/`row_version`. Only a `version` in the natural
# KEY (never seen) falls back to `row_version`.
for r in roots:
r.attrs = [f for f in r.attrs if f.name != "version"]
r.opt = "row_version" if any(f.name == "version" for f in r.key) else "version"
for x in r.descendants:
x.opt = r.opt
# event names: <Entity><ToState>, made unique
used = set()
for st in model.steps:
base = st.ent.cls + ("Decided" if st.decides else "Updated" if st.loop else camel(st.to[0]))
name = base if base not in used else base + "From" + camel(st.frm[0])
n = 2
while name in used:
name, n = f"{base}{n}", n + 1
used.add(name)
st.event = name
model.skipped = skipped
model.configs, model.histories, model.settings_table = configs, histories, settings_table
model.ai_steps = _ai_steps(spec, text)
model.settings = _settings(model, spec)
model.seams = _seams(model, spec)
_resolve_permissions(model)
# stories → the root whose words they use most
for s in spec.get("stories") or []:
if not isinstance(s, dict) or not s.get("id"):
continue
st = Story(str(s["id"]), str(s.get("kind") or "person"), clean(s.get("story")), clean(s.get("given")),
clean(s.get("when")), clean(s.get("then")))
text = " ".join([st.story, st.given, st.when, st.then]).lower()
best, score = None, 0
for r in roots:
sc = sum(text.count(human(x.id)) for x in (r, *r.descendants))
if sc > score:
best, score = r, sc
if best is None and entities:
continue # an --entity run leaves out the stories of other aggregates
st.root = best or next((r for r in roots if r.steps), roots[0] if roots else None)
if st.root is None:
continue
st.root.stories.append(st)
model.stories.append(st)
return model
SEES_ALL = re.compile(r"^\s*(every|everything|all)\b", re.I) # the actor's OWN clause: "their own; a manager sees every" is not
SERVICE = re.compile(r"service account|service client|scheduler", re.I)
def read_all(m: "Model", root_id: str) -> str:
"""0.9.2: seeing every owner's rows is a PERMISSION, never a role name — `<ns>:<root>:read_all`."""
return f"{m.ns}:{root_id}:read_all"
def default_roles(m: "Model") -> dict:
"""The brick's DEFAULT ROLES, one per actor with permissions: the actor's permissions, plus `read_all` on every
owned root when its `sees:` says every/all. A recommendation the composition maps to its own role names — the
brick decides by permission and never reads a role (BRICKS.md, "Permissions")."""
owned = [r.id for r in m.roots if r.has("owner")]
out = {}
for a in m.actors:
perms = [str(p) for p in a.get("permissions") or []]
if not perms:
continue
if SEES_ALL.search(str(a.get("sees") or "").split(";")[0]):
perms += [read_all(m, r) for r in owned if read_all(m, r) not in perms]
out[str(a["id"])] = perms
return out
def service_actors(m: "Model") -> list:
"""Actors that are machines (a scheduler, a service account): `service_clients` in roles.yaml."""
return [str(a["id"]) for a in m.actors if a.get("permissions") and
SERVICE.search(f"{a.get('name') or ''} {a.get('does') or ''} {a.get('sees') or ''}")]
def _resolve_permissions(m: Model) -> None:
"""Each command's permission from the catalog: the acting actors' grant on the entity whose verb the transition
names, else `<ns>:<entity>:write`, else any write the actors hold on it — and only then a made-up string,
flagged — 0.9.1: never a real-looking string: what the catalog does not name is
`TODO(permission):<what it would be>`, a permission no one holds (fail closed) until the spec names it."""
held = {str(a["id"]): [str(p) for p in a.get("permissions") or []] for a in m.actors}
cat = [p for p in m.catalog if p.startswith(m.ns + ":")]
def objects(e: Ent) -> list:
out = [e.id, e.root.id]
if "_" in e.id:
out.append(e.id.split("_", 1)[1])
return list(dict.fromkeys(out))
def pick(e: Ent, by: list, hints: list) -> tuple:
mine = [p for a in by for p in held.get(a, []) if p in cat] or cat
for obj in objects(e):
for p in mine:
_, o, v = (p.split(":") + ["", ""])[:3]
if o == obj and v in hints:
return p, False
acting = [p for a in by for p in held.get(a, []) if p in cat]
for obj in objects(e):
for p in acting:
_, o, v = (p.split(":") + ["", ""])[:3]
if o == obj and v not in ("read", "read_all"):
return p, False
for obj in objects(e):
if f"{m.ns}:{obj}:write" in cat:
return f"{m.ns}:{obj}:write", False
for obj in objects(e):
for p in mine:
_, o, v = (p.split(":") + ["", ""])[:3]
if o == obj and v not in ("read", "read_all"):
return p, False
return f"{TODO_PERM}{m.ns}:{e.root.id}:write", True
for st in m.steps:
hints = [w for w in re.split(r"_", st.key) if w] + [slug(s) for s in st.to] + slug(st.task).split("_")
st.permission, st.permission_todo = pick(st.ent, st.by, hints)
if st.permission_todo:
m.todo_perms.add(st.permission)
for r in m.roots:
rp = next((f"{m.ns}:{o}:read" for o in objects(r) if f"{m.ns}:{o}:read" in cat), "")
m.read_perm[r.id] = rp
wp, todo = pick(r, [], ["write", "keep"])
m.write_perm[r.id] = wp
if todo and not any(s.creates and s.ent is r for s in r.steps):
m.todo_perms.add(wp)
if m.config_rows or m.configs:
sp = next((p for p in (f"{m.ns}:settings:write", f"{m.ns}:config:write", f"{m.ns}:setting:write")
if p in cat), "")
m.write_perm["__settings__"] = sp or f"{TODO_PERM}{m.ns}:settings:write"
if not sp:
m.todo_perms.add(m.write_perm["__settings__"])
# ── emitters ──────────────────────────────────────────────────────────────────────────────
HEADER = "Generated by `erp-compose scaffold` from module.yaml"
def _imports(need: set, pkg_lines: list = ()) -> str:
std = []
if "dataclass" in need or "field" in need:
std.append("from dataclasses import " + ", ".join(n for n in ("dataclass", "field") if n in need))
dt = [n for n in ("date", "datetime") if n in need]
if dt:
std.append("from datetime import " + ", ".join(dt))
if "Decimal" in need:
std.append("from decimal import Decimal")
if "Enum" in need:
std.append("from enum import Enum")
if "Optional" in need:
std.append("from typing import Optional")
out = "\n".join(std)
if pkg_lines:
out += ("\n\n" if out else "") + "\n".join(pkg_lines)
return out + "\n" if out else ""
def _ann(f: Field, need: set, optional: bool = False) -> str:
t = f.py
if t in ("Decimal", "date", "datetime"):
need.add(t)
if optional or f.nullable:
need.add("Optional")
return f"Optional[{t}]"
return t
def gen_values(m: Model) -> str:
body, need = [], set()
for r in m.roots:
for e in (r, *r.descendants):
enums = ([e.state] if e.state and e.state.enum else []) + [f for f in e.attrs if f.enum]
for f in enums:
need.add("Enum")
what = (f"{e.id}: its lifecycle states (module.yaml workflows)" if f is e.state
else f"{e.id}.{f.name}: the values the spec allows")
body.append(f"\n\nclass {f.vo}(str, Enum):\n" + docstring(what, 4)
+ "".join(f" {member(v)} = {lit(v)}\n" for v in f.enum))
head = ('"""Value objects: the states each lifecycle may be in, and the attributes with a fixed set of values.\n\n'
"Stored as their plain strings; naming them here makes a typo a NameError, not a new state.\n"
f'{HEADER}.\n"""\n')
return head + _imports(need) + "".join(body)
def gen_events(m: Model) -> str:
need, body = {"dataclass"}, []
for st in m.steps:
e = st.ent
lines = "".join(f" {f.name}: {_ann(f, need)}\n" for f in e.key)
what = f"{st.key}: {e.id} {', '.join(st.frm)} → {', '.join(st.to)}" + (f". {st.effect}" if st.effect else "")
extra = " outcome: str = \"\"\n" if st.decides else ""
body.append(f"\n\n@dataclass(frozen=True)\nclass {st.event}:\n" + docstring(what, 4) + lines + extra
+ ' by: str = ""\n')
if not body:
need = set()
head = ('"""Domain events: one per workflow transition, named for what happened. An aggregate records them in\n'
"`.events`; nothing publishes them yet — a brick that needs another to react calls its seam.\n"
f'{HEADER}.\n"""\n')
return head + _imports(need) + "".join(body)
def _dataclass(e: Ent, need: set, children: list) -> str:
need.add("dataclass")
out = f"\n\n@dataclass\nclass {e.cls}:\n" + docstring(e.meaning or f"A {human(e.id)}.", 4)
for f in e.key:
out += f" {f.name}: {_ann(f, need)}\n"
fields_ = list(e.attrs) + ([e.state] if e.state else [])
for f in fields_:
if f.kind == "decimal" or f.kind in ("rate", "pct"):
need.add("Decimal")
out += f" {f.name}: {_ann(f, need)} = {f.default}\n"
for c in children:
need.add("field")
out += f" {c.list_attr}: list = field(default_factory=list)\n"
return out
def _sig_key(fields_: list, need: set) -> str:
return "".join(f", {f.name}: {_ann(f, need)}" for f in fields_)
def gen_domain(m: Model, r: Ent) -> tuple:
need, todos = {"field"}, 0
desc = r.descendants
out = ""
for c in desc:
out += _dataclass(c, need, [])
out += _dataclass(r, need, desc)
out += " events: list = field(default_factory=list, repr=False, compare=False)\n"
out += ("\n def _record(self, event):\n self.events.append(event)\n return event\n")
events = sorted({st.event for x in (r, *desc) for st in x.steps})
for c in desc:
if not any(not s.creates for s in c.steps):
continue
ex = c.extra_key
cond = " and ".join(f"it.{f.name} == {f.name}" for f in ex) or "True"
what = ", ".join("{" + f.name + "}" for f in ex)
out += (f"\n def _{c.id}(self{_sig_key(ex, need)}) -> {c.cls}:\n"
f" for it in self.{c.list_attr}:\n if {cond}:\n return it\n"
f" raise RuleBroken(f\"there is no {human(c.id)} {what}\")\n")
steps = [st for x in (r, *desc) for st in x.steps]
if steps:
out += "\n # ── the lifecycle (module.yaml workflows) " + "─" * 66 + "\n"
for st in steps:
e = st.ent
vo = e.state.vo
req = f", {st.requires}: str" if st.requires else ""
dec = ", outcome: str" if st.decides else ""
doc = f"{st.task}: {', '.join(st.frm)} → {', '.join(st.to)}."
if st.gate:
doc += f" Gate: {st.gate}."
if st.effect:
doc += f" Effect: {st.effect}."
gate = ""
if st.gate:
gate += f" # TODO(rule): the gate — {clean(st.gate)}\n"
todos += 1
if st.requires_text and not st.requires:
gate += f" # TODO(rule): requires — {st.requires_text}\n"
todos += 1
if st.effect:
gate += f" # TODO(rule): the effect — {clean(st.effect)}\n"
todos += 1
reqchk = ""
if st.requires:
reqchk = (f" if not {st.requires}:\n"
f" raise RuleBroken(\"{human(st.requires)} is required\")\n")
if e.has(st.requires) or any(f.name == st.requires for f in e.attrs):
reqchk += f" it.{st.requires} = {st.requires}\n"
ev_args = ", ".join(f"{f.name}=it.{f.name}" for f in e.key) + (", outcome=outcome" if st.decides else "") \
+ ", by=by"
to_vals = ", ".join(f"{vo}.{member(s)}" for s in st.to)
frm_vals = ", ".join(f"{vo}.{member(s)}" for s in st.frm if s != "none")
if st.creates and e is r:
out += (f"\n @classmethod\n def {st.key}(cls{_sig_key(r.key, need)}{dec}{req}, *, by: str = \"\", "
f"**attrs) -> \"{r.cls}\":\n" + docstring(doc + " Returns the new aggregate.", 8))
out += " it = cls(" + ", ".join(f"{f.name}={f.name}" for f in r.key) + ", **attrs)\n"
target = f"{vo}.{member(st.to[0])}.value" if not st.decides else "outcome"
if st.decides:
out += (f" if outcome not in ({to_vals},):\n"
f" raise RuleBroken(f\"a new {human(e.id)} is one of {', '.join(st.to)}, "
f"not {{outcome}}\")\n")
out += gate + reqchk + f" it.{e.state.name} = {target}\n"
out += f" it._record({st.event}({ev_args}))\n return it\n"
continue
ret = f"Optional[{st.event}]"
need.add("Optional")
if st.creates:
ex = e.extra_key
cond = " and ".join(f"x.{f.name} == {f.name}" for f in ex) or "True"
out += (f"\n def {st.key}(self{_sig_key(ex, need)}{dec}{req}, *, by: str = \"\", **attrs) -> {ret}:\n"
+ docstring(doc + " Adding the same one again changes nothing.", 8)
+ f" if any({cond} for x in self.{e.list_attr}):\n return None\n")
ctor = ", ".join([f"{f.name}=self.{f.name}" for f in r.key] + [f"{f.name}={f.name}" for f in ex])
out += f" it = {e.cls}({ctor}, **attrs)\n"
target = f"{vo}.{member(st.to[0])}.value" if not st.decides else "outcome"
out += gate + reqchk + f" it.{e.state.name} = {target}\n"
out += f" self.{e.list_attr}.append(it)\n return self._record({st.event}({ev_args}))\n"
continue
ex = e.extra_key if e is not r else []
out += f"\n def {st.key}(self{_sig_key(ex, need)}{dec}{req}, *, by: str = \"\") -> {ret}:\n" + docstring(doc, 8)
out += (" it = self\n" if e is r else
f" it = self._{e.id}(" + ", ".join(f.name for f in ex) + ")\n")
sname = e.state.name
if st.decides:
out += (f" if outcome not in ({to_vals},):\n"
f" raise RuleBroken(f\"{a_(human(e.id))} decision is one of {', '.join(st.to)}, "
f"not {{outcome}}\")\n"
f" if it.{sname} == outcome:\n return None # decided already\n")
elif not st.loop:
out += (f" if it.{sname} == {vo}.{member(st.to[0])}:\n"
f" return None # already {st.to[0]}: a repeat changes nothing\n")
out += (f" if it.{sname} not in ({frm_vals},):\n"
f" raise RuleBroken(f\"a {{it.{sname}}} {human(e.id)} cannot move to "
f"{' or '.join(st.to)}\")\n")
out += gate + reqchk
if st.decides:
out += f" it.{sname} = outcome\n"
elif not st.loop:
out += f" it.{sname} = {vo}.{member(st.to[0])}.value\n"
out += f" return self._record({st.event}({ev_args}))\n"
if r.stories:
out += "\n # ── invariants: one stub per story; implement it, call it where it guards " + "─" * 31 + "\n"
for s in r.stories:
then = clean(s.then)
out += (f"\n def {s.check}(self) -> None:\n"
+ docstring(f"TODO(rule): {s.id} — Then: {then}", 8)
+ f" raise RuleBroken({lit('TODO(rule) ' + s.id + ': not implemented')})\n")
todos += 1
pkg = [f"from {m.package}.domain.errors import RuleBroken"]
if events:
pkg.append(f"from {m.package}.domain.events import " + ", ".join(events))
vos = sorted({x.state.vo for x in (r, *desc) if x.state} | {f.vo for x in (r, *desc) for f in x.attrs if f.vo})
if vos:
pkg.append(f"from {m.package}.domain.values import " + ", ".join(vos))
if "RuleBroken" not in out:
pkg = pkg[1:]
head = (f'"""The {r.cls} aggregate — the rules, and nothing else.\n\n'
+ "\n".join(textwrap.wrap(clean(r.meaning or f"A {human(r.id)}."), 108)) + "\n\n"
"No database, no HTTP, no identity: a rule here is true whoever calls it and however it is stored.\n"
f'{HEADER}; the TODO lines are the work left.\n"""\n')
return head + _imports(need, pkg) + out, todos
def gen_errors() -> str:
return ('"""The one refusal a rule makes. The use cases turn it into a 409 with these words, through both doors."""\n'
"\n\nclass RuleBroken(Exception):\n"
' """A business rule said no. The message is written for the person who tried."""\n')
def _col(f: Field, pk: bool) -> str:
return f'Column("{f.name}", {KINDS[f.kind][1]}' + (", primary_key=True)" if pk else ")")
def _cols(e: Ent) -> list:
return [f.name for f in e.fields] + ([e.state.name] if e.state else [])
def gen_tables(m: Model) -> str:
types, out, mapping = {"Column", "MetaData", "Table", "TIMESTAMP", "select"}, [], []
domain_imports = []
hist = m.histories[0] if m.histories else None
for r in m.roots:
parts = (r, *r.descendants)
for e in parts:
fs = [*e.key, *e.attrs] + ([e.state] if e.state else [])
for f in fs:
types.add(KINDS[f.kind][1].split("(")[0])
cols = [_col(f, True) for f in e.key] + [_col(f, False) for f in fs[len(e.key):]]
if e is r:
cols.append(f'Column("{r.opt}", Integer)')
types.add("Integer")
cols += ['Column("created_at", TIMESTAMP)', 'Column("updated_at", TIMESTAMP)']
out.append(f'\n{e.table} = Table("{e.table}", md,\n' + ",\n".join(" " * 4 + c for c in cols) + ")\n")
out.append(f"{e.table.upper()}_COLUMNS = (" + ", ".join(lit(c) for c in _cols(e))
+ ("," if len(_cols(e)) == 1 else "") + ")\n")
agg = (f"\n{r.id.upper()} = Aggregate(" + ",\n ".join(
f"Part({e.table}, (" + ", ".join(lit(f.name) for f in e.key) + ("," if len(e.key) == 1 else "") + "))"
for e in parts) + f',\n version="{r.opt}")\n')
out.append(agg)
domain_imports.append(f"from {m.package}.domain.{r.id} import " + ", ".join(sorted(e.cls for e in parts)))
# the repository
ksig = ", ".join(f"{f.name}" for f in r.key)
kd = "{" + ", ".join(f'"{f.name}": {f.name}' for f in r.key) + "}"
child_load = "".join(
f" agg.{c.list_attr} = [_make({c.cls}, x, {c.table.upper()}_COLUMNS)\n"
f" for x in _sorted(rows[\"{c.table}\"], {c.table.upper()}_COLUMNS[:{len(c.key)}])]\n"
for c in r.descendants)
child_save = "".join(
f',\n "{c.table}": [_pick(x, {c.table.upper()}_COLUMNS) for x in agg.{c.list_attr}]'
for c in r.descendants)
owner = r.has("owner")
sel = ", ".join(f"{r.table}.c.{f.name}" for f in r.key)
one = len(r.key) == 1
mapping.append(
f"\n\nclass {r.cls}Repository:\n"
f' """Loads and saves a {human(r.id)} whole. Takes an open connection: the use case owns the transaction."""\n'
f"\n def __init__(self, conn):\n self.conn = conn\n"
f"\n def get(self, {ksig}) -> Optional[tuple]:\n"
f' """({r.cls}, the version it was loaded at), or None."""\n'
f" rows = {r.id.upper()}.load(self.conn, {kd})\n"
f" if rows is None:\n return None\n"
f" agg = _make({r.cls}, rows[\"{r.table}\"][0], {r.table.upper()}_COLUMNS)\n"
+ child_load
+ f" return agg, {r.id.upper()}.version_of(rows)\n"
f"\n def save(self, agg: {r.cls}, expected_version: int) -> int:\n"
f' """Writes version expected_version + 1, or raises Conflict (a 409 through both doors)."""\n'
+ (f" version = {r.id.upper()}.save(self.conn, {{\n"
f' "{r.table}": [_pick(agg, {r.table.upper()}_COLUMNS)]{child_save}}},\n'
f" expected_version=expected_version)\n"
f" # append-only history ({hist.table}): one row per event this save records, never rewritten\n"
f" _append_history(self.conn, {lit(r.id)}, {kd.replace(': ', ': agg.')}, "
f"{('agg.' + r.state.name) if r.state else 'None'}, agg.events, version or expected_version)\n"
" agg.events.clear()\n return version\n" if hist else
f" return {r.id.upper()}.save(self.conn, {{\n"
f' "{r.table}": [_pick(agg, {r.table.upper()}_COLUMNS)]{child_save}}},\n'
f" expected_version=expected_version)\n")
+ f"\n def ids(self, owner: Optional[str] = None) -> list:\n"
f" stmt = select({sel})\n"
+ (f" if owner is not None:\n stmt = stmt.where({r.table}.c.owner == owner)\n" if owner else "")
+ (" return sorted(row[0] for row in self.conn.execute(stmt))\n" if one else
" return sorted(list(row) for row in self.conn.execute(stmt))\n"))
settings, plain, need_upsert = "", "", False
for e in [*m.configs, *m.histories]:
fs = [*e.key, *e.attrs]
for f in fs:
types.add(KINDS[f.kind][1].split("(")[0])
cols = [_col(f, True) for f in e.key] + [_col(f, False) for f in e.attrs]
cols += ['Column("created_at", TIMESTAMP)', 'Column("updated_at", TIMESTAMP)']
plain += (f'\n{e.table} = Table("{e.table}", md,\n' + ",\n".join(" " * 4 + c for c in cols) + ")\n"
f"{e.table.upper()}_COLUMNS = (" + ", ".join(lit(f.name) for f in fs) + ("," if len(fs) == 1 else "")
+ ")\n")
keyt = "(" + ", ".join(lit(f.name) for f in e.key) + ("," if len(e.key) == 1 else "") + ")"
order = ", ".join(f"{e.table}.c.{f.name}" for f in e.key)
listing = (f"\n def all(self) -> list:\n stmt = select({e.table}).order_by({order})\n"
f" return [{{k: r._mapping[k] for k in {e.table.upper()}_COLUMNS}} "
"for r in self.conn.execute(stmt)]\n")
if e.role == "config":
need_upsert = True
mapping.append(
f"\n\nclass {e.cls}Repository:\n"
f' """Configuration rows ({e.table}): {e.meaning or "configuration"}. Not an aggregate: no lifecycle,\n'
' no version claim — a keyed upsert, edited under the settings permission."""\n'
"\n def __init__(self, conn):\n self.conn = conn\n" + listing
+ "\n def put(self, row: dict) -> None:\n"
f" upsert(self.conn, {e.table}, [{{k: v for k, v in row.items() if k in {e.table.upper()}_COLUMNS}}], "
f"{keyt})\n")
else:
types.add("insert")
mapping.append(
f"\n\nclass {e.cls}Repository:\n"
f' """Append-only history ({e.table}): {e.meaning or "history"}. Written by the aggregates\' saves\n'
' (`_append_history`), never by a use case, never updated."""\n'
"\n def __init__(self, conn):\n self.conn = conn\n" + listing)
if hist:
hk = hist.key[0].name
cols = []
for f in hist.attrs:
n = f.name
if n in ("kind", "event", "event_type", "type", "event_kind"):
cols.append(f'"{n}": type(event).__name__')
elif n in ("actor", "by", "who", "actor_id", "performed_by"):
cols.append(f'"{n}": data.get("by")')
elif n in ("to_state", "new_state", "state", "status"):
cols.append(f'"{n}": state')
elif n in ("detail", "details", "data", "payload", "event_json"):
cols.append(f'"{n}": json.dumps({{"root": root, **data}}, default=str, sort_keys=True)')
elif n in ("subject_id", "subject_ref", "subject"):
cols.append(f'"{n}": ref')
else:
cols.append(f'"{n}": key.get("{n}", data.get("{n}"))')
plain += (
"\n\ndef _append_history(conn, root: str, key: dict, state, events: list, version: int) -> None:\n"
f' """One {hist.table} row per event a save recorded — append-only: a plain INSERT, its key derived from\n'
" the aggregate, its version and the event's place, so one save never appends twice. A column the event\n"
' does not carry is None: TODO(rule) fill it where the brick knows it (e.g. from_state)."""\n'
" if not events:\n return\n"
" ref = \"|\".join(str(v) for v in key.values())\n now = datetime.now(timezone.utc).replace(tzinfo=None)\n"
" rows = []\n for n, event in enumerate(events, 1):\n data = dataclasses.asdict(event)\n"
f' rows.append({{\n "{hk}": f"{{root}}:{{ref}}:v{{version}}:{{n}}",\n'
+ "".join(f" {c},\n" for c in cols) + ' "created_at": now, "updated_at": now})\n'
f" conn.execute(insert({hist.table}), rows)\n")
if m.settings:
types.update({"String"})
need_upsert = True
seed = ",\n".join(f" ({lit(st.key)}, {lit(st.kind)}, {lit(st.changes)}, "
f"({''.join(lit(c) + ', ' for c in st.choices)}))" for st in m.settings)
settings = (f'\n{m.settings_table} = Table("{m.settings_table}", md,\n'
f' Column("setting_key", String, primary_key=True),\n Column("value", String),\n'
f' Column("changes", String),\n'
f' Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))\n'
"\n# module.yaml config_rows: ONE ROW PER SETTING, typed (key, kind, what it changes, a mode's choices).\n"
"# The value is stored as JSON text; NULL = the reference behaviour. TODO(rule): each setting's default.\n"
f"SETTINGS_SEED = (\n{seed},\n)\n"
"SETTING_KINDS = {k: (kind, choices) for k, kind, _, choices in SETTINGS_SEED}\n"
"\n\ndef _encode(kind: str, choices: tuple, value) -> Optional[str]:\n"
" if value is None:\n return None\n"
" if kind == \"bool\" and not isinstance(value, bool):\n raise ValueError(\"true or false\")\n"
" if kind == \"int\" and (isinstance(value, bool) or not isinstance(value, int)):\n"
" raise ValueError(\"a whole number\")\n"
" if kind == \"date\":\n value = date.fromisoformat(str(value)).isoformat()\n"
" if kind == \"mode\" and value not in choices:\n"
" raise ValueError(\"one of \" + \", \".join(choices))\n"
" if kind in (\"str\", \"mode\") and not isinstance(value, str):\n raise ValueError(\"text\")\n"
" return json.dumps(value)\n"
"\n\ndef _decode(raw: Optional[str]):\n return None if raw is None else json.loads(raw)\n")
mapping.append(
"\n\nclass SettingsRepository:\n"
f' """The setting rows ({m.settings_table}): one per setting, typed — int, bool, date, str or an AI mode."""\n'
"\n def __init__(self, conn):\n self.conn = conn\n"
f"\n def all(self) -> list:\n stmt = select({m.settings_table}).order_by({m.settings_table}.c.setting_key)\n"
" return [{\"setting_key\": r._mapping[\"setting_key\"], \"value\": _decode(r._mapping[\"value\"]),\n"
" \"kind\": SETTING_KINDS.get(r._mapping[\"setting_key\"], (\"str\", ()))[0],\n"
" \"changes\": r._mapping[\"changes\"]} for r in self.conn.execute(stmt)]\n"
"\n def get(self, setting_key: str) -> Optional[dict]:\n"
" return next((s for s in self.all() if s[\"setting_key\"] == setting_key), None)\n"
"\n def set(self, setting_key: str, value, changes: str) -> None:\n"
" \"\"\"Raises ValueError when `value` is not of the setting's kind (None puts the reference back).\"\"\"\n"
" kind, choices = SETTING_KINDS.get(setting_key, (\"str\", ()))\n"
f" upsert(self.conn, {m.settings_table}, [{{\"setting_key\": setting_key, "
"\"value\": _encode(kind, choices, value),\n"
" \"changes\": changes}], (\"setting_key\",))\n"
"\n def seed(self) -> None:\n"
' """A row for every setting the customer has not got yet; never touches one they have set."""\n'
f" have = {{r[0] for r in self.conn.execute(select({m.settings_table}.c.setting_key))}}\n"
" for key, _kind, changes, _choices in SETTINGS_SEED:\n"
" if key not in have:\n self.set(key, None, changes)\n")
facade = ("\n\nclass Repository:\n"
' """Every aggregate\'s repository on one connection: what `use_cases.repository()` yields."""\n'
"\n def __init__(self, conn):\n self.conn = conn\n"
+ "".join(f" self.{plural(r.id)} = {r.cls}Repository(conn)\n" for r in m.roots)
+ "".join(f" self.{plural(e.id)} = {e.cls}Repository(conn)\n" for e in [*m.configs, *m.histories])
+ (" self.settings = SettingsRepository(conn)\n" if m.settings else "")
+ ("\n def seed(self) -> None:\n \"\"\"The rows a new bucket starts with: the settings.\"\"\"\n"
" self.settings.seed()\n" if m.settings else ""))
helpers = ("\n\ndef _pick(obj, names: tuple) -> dict:\n return {n: getattr(obj, n) for n in names}\n"
"\n\ndef _make(cls, row: dict, names: tuple):\n return cls(**{n: row[n] for n in names})\n"
"\n\ndef _sorted(rows: list, key: tuple) -> list:\n"
" return sorted(rows, key=lambda r: tuple(str(r[k]) for k in key))\n")
order = ["BigInteger", "Boolean", "Column", "Date", "Integer", "MetaData", "Numeric", "String", "Table",
"TIMESTAMP", "insert", "select"]
dt_names = [x for x in ("date", "datetime", "timezone") if (x == "date" and m.settings) or (x != "date" and hist)]
stdlib = (["import dataclasses"] if hist else []) + (["import json"] if hist or m.settings else []) + (
[f"from datetime import {', '.join(dt_names)}"] if dt_names else [])
head = ('"""The tables, and one repository per aggregate that loads and saves it whole.\n\n'
"Every table is prefixed, has a natural primary key (never SERIAL) and, for a part, a key that starts with\n"
"its root's — the repository finds stale rows under a root by that prefix. Money is NUMERIC(18,2).\n"
"schema.sql is generated from the same column model; tests/test_schema.py keeps the two equal.\n"
f'{HEADER}.\n"""\n'
+ "".join(x + "\n" for x in stdlib)
+ "from typing import Optional\n\n"
f"from sqlalchemy import {', '.join(t for t in order if t in types)}\n\n"
"from erp_platform.repository import Aggregate, Part" + (", upsert" if need_upsert else "") + "\n\n"
+ "\n".join(domain_imports) + "\n"
"\nmd = MetaData()\n")
return head + "".join(out) + plain + settings + helpers + "".join(mapping) + facade
def _setting_key(m: Model, c: dict) -> str:
t = slug(c.get("table") or c.get("key") or "setting")
return t[len(m.prefix):] if t.startswith(m.prefix) else t
def _ddl_cols(e: Ent, r: Optional[Ent]) -> list:
"""r None: a configuration or history table — no optimistic version column."""
fs = [*e.key, *e.attrs] + ([e.state] if e.state else [])
width = max(len(f.name) for f in fs + [Field("created_at")] + ([Field(r.opt)] if r else []))
rows = [f" {f.name.ljust(width)} {KINDS[f.kind][2]}" + (" NOT NULL" if i < len(e.key) else "")
for i, f in enumerate(fs)]
if e is r:
rows.append(f" {r.opt.ljust(width)} INTEGER")
rows += [f" {'created_at'.ljust(width)} TIMESTAMP", f" {'updated_at'.ljust(width)} TIMESTAMP"]
return rows
def gen_schema(m: Model) -> tuple:
"""(schema.sql, [table names in creation order])."""
out = ["-- " + HEADER + f" ({m.id}). Every table prefixed `{m.prefix}`, a natural primary key each (never\n"
"-- SERIAL), money NUMERIC(18,2), creation columns last so a later column is an append. Additive only.\n"]
names = []
for r in m.roots:
for e in (r, *r.descendants):
cols = _ddl_cols(e, r) + [f" PRIMARY KEY ({', '.join(f.name for f in e.key)})"]
out.append(f"\nCREATE TABLE IF NOT EXISTS {e.table} (\n" + ",\n".join(cols) + "\n);\n")
names.append(e.table)
v = f"{r.table}_versions"
kw = max(len(f.name) for f in r.key + [Field("saved_at"), Field(r.opt)])
cols = [f" {f.name.ljust(kw)} {KINDS[f.kind][2]} NOT NULL" for f in r.key]
cols += [f" {r.opt.ljust(kw)} INTEGER NOT NULL", f" {'saved_at'.ljust(kw)} TIMESTAMP NOT NULL",
f" PRIMARY KEY ({', '.join([f.name for f in r.key] + [r.opt])})"]
out.append("\n-- Optimistic versions (erp_platform.repository): a save CLAIMS version n+1 with a plain INSERT "
"here,\n-- in the same transaction as its rows; a racer from the same version fails 23505 -> 409.\n"
f"CREATE TABLE IF NOT EXISTS {v} (\n" + ",\n".join(cols) + "\n);\n")
names.append(v)
for e in [*m.configs, *m.histories]:
what = ("configuration rows (module.yaml config_rows), edited under the settings permission"
if e.role == "config" else "append-only history: rows are appended on save, never updated")
cols = _ddl_cols(e, None) + [f" PRIMARY KEY ({', '.join(f.name for f in e.key)})"]
out.append(f"\n-- {what}.\nCREATE TABLE IF NOT EXISTS {e.table} (\n" + ",\n".join(cols) + "\n);\n")
names.append(e.table)
if m.settings:
out.append(f"\n-- module.yaml config_rows: one row per setting (JSON value); NULL = the reference behaviour.\n"
f"CREATE TABLE IF NOT EXISTS {m.settings_table} (\n setting_key VARCHAR NOT NULL,\n"
" value VARCHAR,\n changes VARCHAR,\n created_at TIMESTAMP,\n"
" updated_at TIMESTAMP,\n PRIMARY KEY (setting_key)\n);\n")
names.append(m.settings_table)
return "".join(out), names
def gen_migrations(m: Model, schema: str, names: list) -> tuple:
journal = f"{m.prefix}schema_migrations"
seed = "".join(f"INSERT INTO {m.settings_table} (setting_key, value, changes) VALUES ("
f"'{st.key}', NULL, '{st.changes.replace(chr(39), chr(39) * 2)}');\n"
for st in m.settings)
up = (f"-- 00001_init — the brick's first schema ({HEADER}).\n"
f"-- The versioned layout (migrations/NNNNN_<name>/{{up,down}}.sql) with a per-brick journal, {journal}.\n"
"-- erp-platform has no migrator for this layout yet: today `erp-migrate --schema schema.sql` applies the\n"
"-- same tables (schema.sql), and these files are what the migrator will run.\n"
f"\nCREATE TABLE IF NOT EXISTS {journal} (\n migration VARCHAR NOT NULL,\n"
" applied_at TIMESTAMP NOT NULL,\n PRIMARY KEY (migration)\n);\n"
+ schema.split("\n", 2)[2] + ("\n" + seed if seed else "")
+ f"\nINSERT INTO {journal} (migration, applied_at) VALUES ('00001_init', CURRENT_TIMESTAMP);\n")
down = ("-- 00001_init down: drops every table the brick made. Destroys the brick's data — a rollback of a\n"
"-- first install only.\n" + "".join(f"DROP TABLE IF EXISTS {t};\n" for t in [*reversed(names), journal]))
return up, down
def gen_process(m: Model) -> str:
spec = m.spec
data = {"id": m.id, "name": m.name, "version": spec.get("version", 1),
"actors": [{k: a[k] for k in ("id", "name", "external") if k in a} for a in m.actors],
"workflows": spec.get("workflows") or []}
return ("# The brick's process — module.yaml's workflows, copied (pattern.md §15). BRICK.declare_process serves\n"
"# it as BPMN 2.0 at GET /process/bpmn; it is identical to `erp-compose bpmn --from module.yaml`.\n"
"# A transition's key (or its derived <entity>_<from>_to_<to>) is frozen once signed: it is the BPMN id.\n"
+ yaml.safe_dump(data, sort_keys=False, allow_unicode=True, width=120))
def _param(f: Field, need: set, optional: bool) -> str:
return f"{f.name}: {_ann(f, need, optional)}" + (" = None" if optional else "")
def _route_key(r: Ent) -> str:
return "/".join("{" + f.name + "}" for f in r.key)
def gen_use_cases(m: Model) -> tuple:
need = {"Optional"}
B = m.const
out, todos, routes, names = [], 0, set(), set()
uc_list = [] # (name, route, kind, permission, root id, sample-able create?)
def route(method: str, path: str, key: str) -> str:
r = f"{method} {path}"
if r in routes:
r = f"{method} {path}/{key.replace('_', '-')}"
routes.add(r)
return r
def decorator(kind, rt, perm, *, idem=False, if_match=None, destructive=False) -> str:
args = [f'kind="{kind}"', f'route="{rt}"']
if perm:
args.append(f'permission="{perm}"')
if idem:
args.append("idempotent=True")
if destructive:
args.append("destructive=True")
if if_match:
args.append(f'if_match="{if_match}"')
return f"\n\n@{B}.use_case(" + ", ".join(args) + ")\n"
for r in m.roots:
base = "/" + plural(r.id).replace("_", "-")
one = f"{base}/{_route_key(r)}"
keys = ", ".join(f.name for f in r.key)
ksig = ", ".join(f"{f.name}: {_ann(f, need)}" for f in r.key)
what = " ".join("{" + f.name + "}" for f in r.key)
owner = r.has("owner")
rp = m.read_perm.get(r.id) or None
vis = (f"\n\ndef _visible_{r.id}(repo, caller: dict, {ksig}) -> tuple:\n"
f" found = repo.{plural(r.id)}.get({keys})\n")
if owner:
vis += (f" scope = auth.owner_scope(caller, see_all={lit(read_all(m, r.id))})\n"
" if found is None or (scope is not None and found[0].owner != scope):\n")
else:
vis += (" # TODO(visibility): the spec's row_visibility rule decides who sees which; today everyone with\n"
" # the read permission sees every row.\n if found is None:\n")
todos += 1
vis += f" raise Refused(404, f\"no {human(r.id)} {what}\")\n return found\n"
out.append(vis)
# queries
names.update({f"list_{plural(r.id)}", f"get_{r.id}"})
rt = route("GET", base, "list")
out.append(decorator("query", rt, rp)
+ f"def list_{plural(r.id)}(caller: dict" + (", mine: bool = False" if owner else "") + ") -> dict:\n"
+ docstring(f"The {human(plural(r.id))} the caller may see" + (
"; `mine` narrows to the caller's own." if owner else ".") + " Ids only; get one for the rest.", 4)
+ " with repository() as repo:\n"
+ (f" return {{\"{plural(r.id)}\": repo.{plural(r.id)}.ids(auth.owner_scope(caller, mine, "
f"see_all={lit(read_all(m, r.id))}))}}\n"
if owner else f" return {{\"{plural(r.id)}\": repo.{plural(r.id)}.ids()}}\n"))
uc_list.append((f"list_{plural(r.id)}", rt, "query", rp, r.id, None))
rt = route("GET", one, "get")
out.append(decorator("query", rt, rp) + f"def get_{r.id}(caller: dict, {ksig}) -> dict:\n"
+ docstring(f"One {human(r.id)}, whole, with its version, if the caller may see it.", 4)
+ " with repository() as repo:\n"
+ f" return _view(*_visible_{r.id}(repo, caller, {keys}), key=\"{r.opt}\")\n")
uc_list.append((f"get_{r.id}", rt, "query", rp, r.id, None))
attrs = [f for f in r.attrs if f.name != "owner"]
aparams = "".join(", " + _param(f, need, True) for f in attrs)
agiven = ", ".join(f"{f.name}={f.name}" for f in attrs)
own = ", owner=_who(caller)" if owner else ""
root_creates = [st for st in r.steps if st.creates]
if not root_creates:
wp = m.write_perm[r.id]
rt = route("POST", base, "keep")
names.add(f"keep_{r.id}")
out.append(
decorator("command", rt, wp, idem=True, if_match=r.opt)
+ f"def keep_{r.id}(caller: dict, {ksig}, {r.opt}: Optional[int] = None{aparams}) -> dict:\n"
+ docstring(f"Create or change a {human(r.id)}: {r.meaning or ''} Fields left out keep their value; "
"saving the same values again changes nothing.", 4)
+ (" # the permission is not in the spec's catalog — add it there.\n" if wp in m.todo_perms else "")
+ " with repository() as repo, _decide():\n"
+ f" found = repo.{plural(r.id)}.get({keys})\n"
+ " if found is None:\n"
+ f" agg, current = {r.cls}({', '.join(f'{f.name}={f.name}' for f in r.key)}{own}), 0\n"
+ " else:\n"
+ f" agg, current = _visible_{r.id}(repo, caller, {keys})\n"
+ f" _same_version(f\"{human(r.id)} {what}\", current, {r.opt})\n"
+ (f" for name, value in _given({agiven}).items():\n setattr(agg, name, value)\n"
if attrs else "")
+ f" # TODO(rule): what makes a {human(r.id)} valid (the spec's stories for it)\n"
+ f" return _view(agg, repo.{plural(r.id)}.save(agg, current), key=\"{r.opt}\")\n")
todos += 1 + (wp in m.todo_perms)
uc_list.append((f"keep_{r.id}", rt, "command", wp, r.id, "keep"))
for st in [s for x in (r, *r.descendants) for s in x.steps]:
e = st.ent
dec_p = ", outcome: str" if st.decides else ""
req_p = f", {st.requires}: str" if st.requires else ""
dec_a = ", outcome=outcome" if st.decides else ""
req_a = f", {st.requires}={st.requires}" if st.requires else ""
doc = f"{st.task} ({e.id}: {', '.join(st.frm)} → {', '.join(st.to)})."
if st.gate:
doc += f" Only when: {st.gate}."
if st.effect:
doc += f" Then: {st.effect}."
tperm = (" # the permission is not in the spec's catalog — add it there.\n" if st.permission_todo else "")
todos += bool(st.permission_todo)
if st.creates and e is r:
rt = route("POST", base, st.key)
eattrs = [f for f in e.attrs if f.name != "owner"]
out.append(
decorator("command", rt, st.permission, idem=True)
+ f"def {st.key}(caller: dict, {ksig}{dec_p}{req_p}{aparams}) -> dict:\n"
+ docstring(doc + " Making the same one again returns it.", 4) + tperm
+ " with repository() as repo, _decide():\n"
+ f" found = repo.{plural(r.id)}.get({keys})\n"
+ " if found is not None:\n"
+ f" return _view(*_visible_{r.id}(repo, caller, {keys}), key=\"{r.opt}\")\n"
+ f" agg = {r.cls}.{st.key}({keys}{dec_a}{req_a}, by=_who(caller){own}"
+ (f", **_given({', '.join(f'{f.name}={f.name}' for f in eattrs)})" if eattrs else "") + ")\n"
+ f" return _view(agg, repo.{plural(r.id)}.save(agg, 0), key=\"{r.opt}\")\n")
uc_list.append((st.key, rt, "command", st.permission, r.id, st.key))
names.add(st.key)
continue
ex = e.extra_key if e is not r else []
if st.creates:
rt = route("POST", f"{one}/{plural(e.id).replace('_', '-')}", st.key)
exp = "".join(f", {f.name}: {_ann(f, need)}" for f in ex)
cattrs = [f for f in e.attrs if f.name not in ("owner",)]
cparams = "".join(", " + _param(f, need, True) for f in cattrs)
call = (f"agg.{st.key}({', '.join(f.name for f in ex)}{dec_a}{req_a}, by=_who(caller)"
+ (f", **_given({', '.join(f'{f.name}={f.name}' for f in cattrs)})" if cattrs else "") + ")")
idem = True
else:
seg = "/".join([f"{plural(e.id).replace('_', '-')}"] + ["{" + f.name + "}" for f in ex]) if ex else ""
rt = route("POST", f"{one}/" + (seg + "/" if seg else "") + st.key.replace("_", "-"), st.key)
exp = "".join(f", {f.name}: {_ann(f, need)}" for f in ex)
cparams = ""
args = [f.name for f in ex] + (["outcome=outcome"] if st.decides else []) \
+ ([f"{st.requires}={st.requires}"] if st.requires else []) + ["by=_who(caller)"]
call = f"agg.{st.key}({', '.join(args)})"
idem = not st.loop
out.append(
decorator("command", rt, st.permission, idem=idem, if_match=r.opt)
+ f"def {st.key}(caller: dict, {ksig}{exp}{dec_p}{req_p}, {r.opt}: Optional[int] = None{cparams}) -> dict:\n"
+ docstring(doc + (" A repeat changes nothing." if idem else ""), 4) + tperm
+ " with repository() as repo, _decide():\n"
+ f" agg, current = _visible_{r.id}(repo, caller, {keys})\n"
+ f" _same_version(f\"{human(r.id)} {what}\", current, {r.opt})\n"
+ f" if {call} is None:\n"
+ f" return _view(agg, current, key=\"{r.opt}\")\n"
+ f" return _view(agg, repo.{plural(r.id)}.save(agg, current), key=\"{r.opt}\")\n")
uc_list.append((st.key, rt, "command", st.permission, r.id, None))
names.add(st.key)
sp = m.write_perm.get("__settings__", "")
perm_note = " # the permission is not in the spec's catalog — add it there.\n" if sp in m.todo_perms else ""
if m.settings:
need.add("Union")
rt1, rt2 = route("GET", "/settings", "list"), route("POST", "/settings/{setting_key}", "set")
out.append(decorator("query", rt1, None) + f"def list_{slug(m.ns)}_settings(caller: dict) -> dict:\n"
+ docstring("The settings the customer edits (module.yaml config_rows), one row per setting with "
"its kind; a value of null is the reference behaviour.", 4)
+ " with repository() as repo:\n return {\"settings\": repo.settings.all()}\n")
out.append(decorator("command", rt2, sp, idem=True)
+ f"def set_{slug(m.ns)}_setting(caller: dict, setting_key: str, "
"value: Optional[Union[bool, int, str]] = None) -> dict:\n"
+ docstring("Change one setting, typed by its kind (a whole number, true/false, an ISO date, text "
"or one of an AI step's modes). null puts the reference behaviour back. Setting the "
"same value again changes nothing.", 4)
+ perm_note
+ " with repository() as repo:\n row = repo.settings.get(setting_key)\n"
+ " if row is None:\n raise Refused(404, f\"no setting {setting_key}\")\n"
+ " # TODO(rule): what else makes a value valid for this setting (a range, a list's words)\n"
+ " try:\n repo.settings.set(setting_key, value, row[\"changes\"])\n"
+ " except ValueError as e:\n"
+ " raise Refused(422, f\"{setting_key} is {e}\") from None\n"
+ " return {\"setting_key\": setting_key, \"value\": value}\n")
uc_list.append((f"list_{slug(m.ns)}_settings", rt1, "query", None, None, None))
uc_list.append((f"set_{slug(m.ns)}_setting", rt2, "command", sp, None, None))
for e in m.configs:
base = "/" + plural(e.id).replace("_", "-")
rt1 = route("GET", base, "list")
rt2 = route("POST", base + "/" + "/".join("{" + f.name + "}" for f in e.key), "set")
ksig = ", ".join(f"{f.name}: {_ann(f, need)}" for f in e.key)
aparams = "".join(", " + _param(f, need, True) for f in e.attrs)
given = ", ".join(f"{f.name}={f.name}" for f in e.attrs)
keyd = "{" + ", ".join(f'"{f.name}": {f.name}' for f in e.key) + "}"
out.append(decorator("query", rt1, None) + f"def list_{plural(e.id)}(caller: dict) -> dict:\n"
+ docstring(f"Configuration: {e.meaning or human(plural(e.id))}.", 4)
+ f" with repository() as repo:\n return {{\"{plural(e.id)}\": repo.{plural(e.id)}.all()}}\n")
out.append(decorator("command", rt2, sp, idem=True)
+ f"def set_{e.id}(caller: dict, {ksig}{aparams}) -> dict:\n"
+ docstring(f"Change one {human(e.id)} row (configuration — the settings permission, not a "
"lifecycle). Fields left out keep their value; the same values again change nothing.", 4)
+ perm_note
+ " with repository() as repo:\n"
+ f" row = next((x for x in repo.{plural(e.id)}.all() if all(x[k] == v for k, v in "
f"{keyd}.items())), {keyd})\n"
+ (f" row.update(_given({given}))\n" if given else "")
+ f" repo.{plural(e.id)}.put(row)\n return row\n")
uc_list.append((f"list_{plural(e.id)}", rt1, "query", None, None, None))
uc_list.append((f"set_{e.id}", rt2, "command", sp, None, None))
for e in m.histories:
out.append(f"\n\n# {e.table}: append-only history, appended by the aggregates' saves — no command writes it.\n"
"# Reading it is a query you add with its row visibility (TODO(visibility)).\n")
rule = m.row_rule or "not stated in the spec — ask the customer and record it"
domain_imports = [f"from {m.package}.domain.errors import RuleBroken"] + [
f"from {m.package}.domain.{r.id} import {r.cls}" for r in m.roots]
head = ('"""The use cases — what a person (or their agent) can do here.\n\n'
"Each is declared ONCE and served as an API route and as an MCP tool. Two things are decided here and\n"
"nowhere else: WHO MAY (the `permission` on the declaration) and WHICH ROWS (`_visible_<root>()`, from the\n"
"caller: a row the caller may not see is a 404, never a 403). Each command is one unit of work: load the\n"
"aggregate, let the domain decide, save it whole at the version it was loaded (`If-Match` for a screen).\n\n"
+ "\n".join(textwrap.wrap("Row visibility (module.yaml): " + rule, 112)) + "\n\n"
f'{HEADER}.\n"""\n'
"import dataclasses\nfrom contextlib import contextmanager\n"
+ (f"from datetime import {', '.join(n for n in ('date', 'datetime') if n in need)}\n"
if need & {"date", "datetime"} else "")
+ ("from decimal import Decimal\n" if "Decimal" in need else "")
+ f"from typing import {', '.join(x for x in ('Optional', 'Union') if x in need)}\n\n"
+ ("from erp_platform import auth\n" if any(r.has("owner") for r in m.roots) else "")
+ "from erp_platform import repository as repository_\n"
+ "from erp_platform.bricks import Brick, Refused\nfrom sqlalchemy.exc import IntegrityError\n\n"
+ "\n".join(domain_imports) + "\n\n"
+ f"{B} = Brick({lit(m.ns)}, version=\"0.1.0\",\n"
+ f"{' ' * (len(B) + 9)}instructions={lit(m.summary or m.name)})\n"
+ (f"# The process view (BRICKS.md): the workflows as BPMN, GET /process/bpmn.\n{B}.declare_process(\"process.yaml\")\n"
if m.has_process else "# No workflows in the spec: a master-data brick, waived in compose.yaml "
"(`process_view: none # master data only`).\n")
+ "\n\n@contextmanager\ndef _repository_default():\n"
+ " from erp_platform import db # imported late: it needs BUCKET\n"
+ " with _on_engine(db.engine) as repo:\n yield repo\n"
+ "\n\ndef _lost_a_unique_race(exc: IntegrityError) -> bool:\n"
+ ' """A duplicate key — but not a version claim, which the doors already answer as \'changed since you '
'opened it\'."""\n'
+ " if repository_.as_conflict(exc) is not None:\n return False\n"
+ " text = f\"{getattr(exc, 'orig', '')} {exc}\"\n"
+ ' return "23505" in text or "duplicate key" in text or "UNIQUE constraint" in text\n'
+ "\n\n@contextmanager\ndef _on_engine(engine):\n"
+ ' """One transaction. On DataK3 a racer\'s duplicate key (two people creating the same natural key at '
'once) can\n'
+ " surface at COMMIT, after the use case has returned — so it is caught here, around the commit, and "
"answered as\n"
+ ' the retry-safe 409 a conflict seen mid-use-case gets, never a 500."""\n'
+ f" from {m.package}.infrastructure.tables import Repository\n"
+ " try:\n with engine.begin() as conn:\n yield Repository(conn)\n"
+ " except IntegrityError as exc:\n if not _lost_a_unique_race(exc):\n raise\n"
+ ' raise Refused(409, "Someone else wrote the same record a moment ago - load it again and retry") '
"from None\n"
+ "\n\nrepository = _repository_default # tests swap in one on SQLite\n"
+ "\n\n@contextmanager\ndef _decide():\n"
+ ' """A broken business rule is a 409 with the rule\'s own words, through both doors."""\n'
+ " try:\n yield\n except RuleBroken as broken:\n raise Refused(409, str(broken)) from None\n"
+ "\n\ndef _who(caller: dict) -> str:\n return caller.get(\"email\") or caller.get(\"sub\") or \"\"\n"
+ "\n\ndef _given(**values) -> dict:\n"
+ ' """The inputs the caller sent: None means "left out", never "set to nothing"."""\n'
+ " return {k: v for k, v in values.items() if v is not None}\n"
+ "\n\ndef _same_version(what: str, current: int, asked: Optional[int]) -> None:\n"
+ " if asked is not None and int(asked) != current:\n"
+ " raise Refused(409, f\"{what} changed since you loaded it (you have version {asked}, it is at \"\n"
+ " f\"{current}) — load it again\")\n"
+ "\n\ndef _view(agg, version: int, key: str = \"version\") -> dict:\n"
+ " out = {}\n for f in dataclasses.fields(agg):\n if f.name == \"events\":\n continue\n"
+ " v = getattr(agg, f.name)\n"
+ " out[f.name] = [dataclasses.asdict(x) for x in v] if isinstance(v, list) else v\n"
+ " out[key] = version\n return out\n")
return head + "".join(out), todos, uc_list
def _ai_capability(a: dict) -> str:
"""The AI-step capability (capabilities.yaml 0.5) a step's words describe, when the spec names none: words a
person sends → ai.draft_message; a warning → ai.flag_risk; anything else proposes values → ai.suggest."""
words = " ".join(clean(a.get(k)) for k in ("key", "step", "output")).lower().replace("_", " ")
if re.search(r"\b(?:message|reminder|e-?mail|reply|letter|cover note)\b", words):
return "ai.draft_message"
if re.search(r"\b(?:flags?|risks?|shortfall|warns?|warning|anomal\w*)\b", words):
return "ai.flag_risk"
return "ai.suggest"
def gen_ai(m: Model) -> tuple:
"""The spec's ai_steps (key, benefit, mode, modes_allowed, safe_for_auto, fallback, eval; capability optional)."""
from erp_platform import capabilities
rows, seams, todos, known = [], [], 0, False
for a in m.ai_steps:
if not isinstance(a, dict):
continue
step = slug(a.get("key") or a.get("step") or "step")
cap = str(a.get("capability") or "") or _ai_capability(a)
try:
ok = bool(cap) and capabilities.contract().get(cap) is not None
except Exception: # noqa: BLE001 — unknown: a TODO, never invented API
ok = False
allowed = [str(x) for x in a.get("modes_allowed") or ["off", "suggest"]]
mode = str(a.get("mode") or "suggest")
safe = bool(a.get("safe_for_auto", a.get("safe", False)))
rows.append(f" {lit(step)}: {{\n"
f" \"capability\": {lit(cap) if cap else 'None'},\n"
f" \"mode\": {lit(mode)},\n"
f" \"modes_allowed\": [{', '.join(lit(x) for x in allowed)}],\n"
f" \"safe_for_auto\": {safe},\n"
f" \"fallback\": {lit(clean(a.get('fallback')) or 'a manual task for a person')},\n"
f" \"benefit\": {lit(clean(a.get('benefit')))},\n"
f" \"inputs\": [{', '.join(lit(clean(x)) for x in a.get('inputs') or [])}],\n"
f" \"eval\": {lit(clean(a.get('eval')))},\n }},\n")
if ok:
known = True
seams.append(f" {lit(step)}: CapabilitySeam({lit(cap)}),\n")
else:
why = (f"{lit(cap)} is not in erp-platform's capabilities.yaml" if cap
else "no capability is named for it")
seams.append(f" # TODO(ai): {step}: {why} — propose its provider capability there; until then the\n"
f" # step is its fallback (a manual task).\n")
todos += 1
todos += 1
text = ('"""AI steps (erp-roadmap #53, the AI-step contract): an AI step is a PROVIDER of a capability, with the\n'
"ManualTask as its fallback and its reviewer. The mode is part of the signed BPMN: off | suggest (the\n"
"default: AI drafts, a person approves) | auto-with-review | auto (only a step marked safe_for_auto; never\n"
"destructive or money-moving without approval).\n\n"
"TODO(ai): erp-platform has no step-mode API yet. Until it lands, `suggest` is this brick's job: the use\n"
"case stores the provider's draft as a ManualTask for a person to approve, and never acts on it alone.\n"
f'{HEADER}.\n"""\n'
+ ("from erp_platform.capabilities import CapabilitySeam\n" if known else "")
+ "\n# What every AI action leaves behind (the AI-step contract): the provider's action log holds the model,\n"
"# the inputs, the output and the confidence; the brick records the approver — who accepted the ManualTask,\n"
"# or reviewed an auto-with-review result.\nAUDIT = (\"model\", \"inputs\", \"output\", \"confidence\", \"approver\")\n"
+ "\nAI_STEPS = {\n" + "".join(rows) + "}\n\nSEAMS = {\n" + "".join(seams) + "}\n"
+ "\n\ndef propose(step: str, idempotency_key: str, input: dict, mode: str):\n"
' """Ask the step\'s provider for a proposal. `off` → None (the process is the fallback). Otherwise the\n'
" provider's Result, or — no provider, or none wired — the ManualTask with the exact input (rule 8). What\n"
" happens next is the caller's, by the mode: suggest → a ManualTask a person accepts, edits or rejects;\n"
" auto-with-review → apply it and open a review task; auto → apply it (safe_for_auto steps only).\n"
' The provider never applies anything."""\n'
" spec = AI_STEPS[step]\n"
" if mode == \"off\":\n return None\n"
" if mode not in spec[\"modes_allowed\"]:\n"
" raise ValueError(f\"{step}: mode {mode!r} is not allowed ({', '.join(spec['modes_allowed'])})\")\n"
" if mode == \"auto\" and not spec[\"safe_for_auto\"]:\n"
" raise ValueError(f\"{step} is not safe for auto: a person approves it\")\n"
" if step not in SEAMS:\n"
" return None # no capability yet: the fallback, a person (the note above)\n"
" return SEAMS[step].call(idempotency_key, input)\n")
return text, todos
def _sample(f: Field):
if f.kind == "str":
return f"{f.name.upper().replace('_', '-')}-1"
return KINDS[f.kind][4]
def gen_tests(m: Model, uc_list: list, workflows: list) -> dict:
files = {}
pkg = m.package
perms = sorted({p for _, _, _, p, _, _ in uc_list if p} | {p for p in m.catalog if p.startswith(m.ns + ":")})
files["tests/conftest.py"] = (
'"""Shared wiring: the real app with storage on a SQLite file (each test its own), personas that carry\n'
'exactly the permissions given, and the two doors — `api(...)` and `tool(...)`. Generated by erp-compose scaffold."""\n'
"import base64\nimport json\nfrom contextlib import contextmanager\n\nimport anyio\nimport httpx\n"
"import httpx2 # the MCP client's HTTP library\nimport pytest\n"
"from mcp import ClientSession\nfrom mcp.client.streamable_http import streamable_http_client\n"
"from sqlalchemy import create_engine\n\n"
f"from {pkg}.app import app\nfrom {pkg}.application import use_cases\n"
f"from {pkg}.infrastructure.tables import Repository, md\n\n"
"ALL = [" + ", ".join(lit(p) for p in perms) + "]\n\n\n"
"@pytest.fixture(autouse=True)\ndef engine(tmp_path, monkeypatch):\n"
" e = create_engine(f\"sqlite:///{tmp_path / 'bucket.db'}\", connect_args={\"check_same_thread\": False})\n"
" md.create_all(e)\n"
+ (" with e.begin() as conn:\n Repository(conn).seed()\n" if m.settings else "")
+ "\n @contextmanager\n def repo():\n with e.begin() as conn:\n yield Repository(conn)\n"
" monkeypatch.setattr(use_cases, \"repository\", repo)\n yield e\n e.dispose()\n\n\n"
"def person(email, permissions, roles=()):\n"
" payload = base64.urlsafe_b64encode(json.dumps({\"permissions\": list(permissions)}).encode()).decode()\n"
" return {\"X-Dodil-User\": json.dumps({\"sub\": email, \"email\": email, \"app_roles\": list(roles)}),\n"
" \"X-Dodil-User-JWT\": f\"h.{payload.rstrip('=')}.s\"}\n\n\n"
"ALICE = person(\"alice@shop.example\", ALL)\nNOBODY = person(\"nobody@shop.example\", [])\n\n\n"
"async def _call(method, path, headers, body=None):\n"
" transport = httpx.ASGITransport(app=app)\n"
" async with httpx.AsyncClient(transport=transport, base_url=\"http://brick\") as c:\n"
" if method == \"GET\":\n return await c.get(path, headers=headers, params=body)\n"
" return await c.request(method, path, headers=headers, json=body)\n\n\n"
"def api(method, path, headers, body=None):\n r = anyio.run(_call, method, path, headers, body)\n"
" return r.status_code, r.json()\n\n\n"
"async def _mcp(headers, fn):\n"
" async with app.router.lifespan_context(app):\n"
" async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app),\n"
" headers=headers, base_url=\"http://brick\") as http:\n"
" async with streamable_http_client(\"http://brick/mcp\", http_client=http) as (r, w):\n"
" async with ClientSession(r, w) as s:\n await s.initialize()\n"
" return await fn(s)\n\n\n"
"def tool(headers, name, args):\n"
" res = anyio.run(_mcp, headers, lambda s: s.call_tool(name, args))\n"
" text = res.content[0].text\n"
" return (\"refused\", text.split(\": \", 1)[1]) if res.is_error else (\"ok\", json.loads(text))\n\n\n"
"def tools(headers):\n"
" listed = anyio.run(_mcp, headers, lambda s: s.list_tools())\n return sorted(t.name for t in listed.tools)\n")
# test_rules.py
need = set()
body = []
seen = set()
for r in m.roots:
if not r.stories:
continue
args = ", ".join(f"{f.name}={_pylit(_sample(f), f, need)}" for f in r.key)
body.append(f"\n\ndef a_{r.id}() -> {r.cls}:\n return {r.cls}({args})\n")
for s in m.stories:
r = s.root
name = f"test_{slug(s.id)}_{slug(' '.join(s.then.split()[:6]))}"[:79].rstrip("_")
while name in seen:
name += "_"
seen.add(name)
text = f"{s.id} ({s.kind}) — {s.story} Given: {s.given} When: {s.when} Then: {s.then}"
body.append(f"\n\n@pytest.mark.xfail(strict=True, reason=\"TODO rule\")\ndef {name}():\n" + docstring(text, 4)
+ f" a_{r.id}().{s.check}() # TODO(test): the When, then assert the Then\n")
imports = ["import pytest"] if m.stories else []
if "date" in need:
imports.insert(0, "from datetime import date")
dom = [f"from {pkg}.domain.{r.id} import {r.cls}" for r in m.roots if r.stories]
files["tests/test_rules.py"] = (
'"""The business rules, with no database — one test per story (module.yaml stories), each an xfail until\n'
"its rule is written: `pytest -q` shows exactly which stories remain. Implement the rule, rewrite the test\n"
'from its Given/When/Then, and remove the mark (strict: a rule that starts passing must lose its mark)."""\n'
+ ("\n".join(imports) + "\n\n" if imports else "") + "\n".join(dom) + ("\n" if dom else "") + "".join(body)
+ ("" if m.stories else "\n\ndef test_the_spec_has_no_stories_yet():\n assert True\n"))
# test_doors.py
creators, roots_by = [], {r.id: r for r in m.roots}
for name, rt, kind, perm, rid, create in uc_list:
if create and rid:
r = roots_by[rid]
st = next((s for s in r.steps if s.key == create), None)
bodyd = {f.name: _sample(f) for f in r.key}
if st is not None and st.decides:
bodyd["outcome"] = st.to[0]
if st is not None and st.requires:
bodyd[st.requires] = "because"
if all(c[0] is not r for c in creators): # the first way to make a root is the one tested
creators.append((r, name, rt, perm, bodyd))
t = ('"""The use cases through both doors, on SQLite: every use case is a route AND a tool, a create answers the\n'
'same through the API and MCP, a missing row is 404, and no permission means nothing is written."""\n'
"from conftest import ALICE, NOBODY, api, tool, tools\n\n"
"ROUTES = [\n" + "".join(f" ({lit(n)}, {lit(rt)}),\n" for n, rt, *_ in uc_list) + "]\n\n\n"
"def test_every_use_case_is_a_route_and_a_tool():\n"
" status, spec = api(\"GET\", \"/openapi.json\", ALICE)\n assert status == 200\n"
" for name, route in ROUTES:\n method, path = route.split(\" \", 1)\n"
" assert method.lower() in spec[\"paths\"].get(path, {}), route\n"
" assert {n for n, _ in ROUTES} <= set(tools(ALICE))\n"
"\n\ndef test_a_unique_race_lost_at_commit_is_409_and_any_other_integrity_error_is_not(tmp_path):\n"
' """_on_engine: on DataK3 a racer\'s duplicate key can surface at COMMIT, after the use case returned. It is the\n'
' retry-safe 409, never a 500; a NOT NULL or other integrity failure still raises as itself."""\n'
" import pytest\n from sqlalchemy import create_engine, text\n from sqlalchemy.exc import IntegrityError\n"
" from erp_platform.bricks import Refused\n"
f" from {pkg}.application.use_cases import _on_engine\n\n"
" e = create_engine(f\"sqlite:///{tmp_path / 'race.db'}\")\n"
" with e.begin() as c:\n"
" c.execute(text(\"CREATE TABLE claim (k TEXT PRIMARY KEY, v TEXT NOT NULL)\"))\n"
" c.execute(text(\"INSERT INTO claim VALUES ('a', 'x')\"))\n"
" with pytest.raises(Refused) as lost:\n with _on_engine(e) as repo:\n"
" repo.conn.execute(text(\"INSERT INTO claim VALUES ('a', 'y')\"))\n"
" assert lost.value.status == 409\n"
" with pytest.raises(IntegrityError):\n with _on_engine(e) as repo:\n"
" repo.conn.execute(text(\"INSERT INTO claim VALUES ('b', NULL)\"))\n")
for r, name, rt, perm, bodyd in creators:
get_path = "/" + plural(r.id).replace("_", "-") + "/" + "/".join(str(bodyd[f.name]) for f in r.key)
keyargs = "{" + ", ".join(f"{lit(f.name)}: {json.dumps(bodyd[f.name])}" for f in r.key) + "}"
t += (f"\n\ndef test_a_{r.id}_through_both_doors():\n"
f" body = {json.dumps(bodyd)}\n"
f" status, made = api(\"POST\", {lit(rt.split(' ', 1)[1])}, ALICE, body)\n"
f" assert status == 200, made\n"
f" assert api(\"GET\", {lit(get_path)}, ALICE)[1] == made\n"
f" assert tool(ALICE, \"get_{r.id}\", {keyargs}) == (\"ok\", made)\n"
f" assert made[{lit(r.opt)}] == 1\n")
t += (f"\n\ndef test_a_missing_{r.id}_is_not_found():\n"
f" assert api(\"GET\", {lit(get_path)}, ALICE)[0] == 404\n"
f" assert tool(ALICE, \"get_{r.id}\", {keyargs})[0] == \"refused\"\n")
if creators:
r, name, rt, perm, bodyd = creators[0]
get_path = "/" + plural(r.id).replace("_", "-") + "/" + "/".join(str(bodyd[f.name]) for f in r.key)
t += (f"\n\ndef test_without_the_permission_nothing_is_written():\n"
f" status, _ = api(\"POST\", {lit(rt.split(' ', 1)[1])}, NOBODY, {json.dumps(bodyd)})\n"
f" assert status == 403\n"
f" assert api(\"GET\", {lit(get_path)}, ALICE)[0] == 404\n")
if workflows:
t += ("\n\ndef test_the_process_view_shows_the_workflows_as_bpmn():\n"
" status, body = api(\"GET\", \"/process/bpmn\", ALICE)\n"
f" assert status == 200 and body[\"processes\"] == {json.dumps(workflows)}\n")
if m.settings:
kinds = {st.key: st.kind for st in m.settings}
t += (f"\n\ndef test_the_settings_are_seeded_one_row_per_setting_typed():\n"
f" status, body = api(\"GET\", \"/settings\", ALICE)\n"
f" assert status == 200\n"
f" assert {{s[\"setting_key\"]: s[\"kind\"] for s in body[\"settings\"]}} == "
f"{json.dumps(dict(sorted(kinds.items())))}\n"
f" assert all(s[\"value\"] is None for s in body[\"settings\"])\n")
wrong = next(((st.key, "x" if st.kind in ("int", "bool", "date") else 7) for st in m.settings), None)
if wrong:
sp = m.write_perm.get("__settings__", "")
expect = ("403 # its permission is a TODO the catalog lacks: no one holds it (0.9.2: fail closed) — "
"name it in the spec, then this is 422" if sp in m.todo_perms else "422")
t += (f"\n\ndef test_a_setting_of_the_wrong_kind_is_refused():\n"
f" path = \"/settings/{wrong[0]}\"\n"
f" assert api(\"POST\", path, ALICE, {{\"value\": {json.dumps(wrong[1])}}})[0] == {expect}\n")
files["tests/test_doors.py"] = t
files["tests/test_schema.py"] = (
'"""schema.sql and the tables in infrastructure/tables.py are one column model: they must not drift."""\n'
"import pathlib\nimport re\n\n"
f"from {pkg}.infrastructure.tables import md\n\n"
f"SRC = pathlib.Path(__file__).resolve().parents[1] / \"src\" / \"{pkg}\"\n\n\n"
"def expected_columns(sql: str) -> dict:\n"
" \"\"\"table -> [column], as a schema file declares it (erp_platform.migrate's reading, without a BUCKET).\"\"\"\n"
" out = {}\n"
" for name, body in re.findall(r\"CREATE TABLE IF NOT EXISTS (\\w+) \\(\\n(.*?)\\n\\);\", sql, re.S):\n"
" lines = [ln.strip() for ln in body.splitlines()]\n"
" out[name] = [ln.split()[0] for ln in lines if ln and not ln.startswith((\"PRIMARY KEY\", \"--\"))]\n"
" return out\n\n\n"
"def test_schema_sql_is_the_tables():\n"
" declared = expected_columns((SRC / \"schema.sq