dodil/erp-brick-ar
DODIL ERP brick: accounts receivable
scripts/gen_schema.py
"""Write schema.sql and migrations/00001_init/{up,down}.sql from the column model in infrastructure/tables.py.
python scripts/gen_schema.py # rewrite the three files
python scripts/gen_schema.py --check # fail if they differ (tests/test_schema.py does the column comparison)
The model is the source of truth; a later release APPENDS columns (never drop, rename or retype) and adds a new migration.
"""
import pathlib
import sys
from sqlalchemy import TIMESTAMP, BigInteger, Boolean, Date, Integer, Numeric, String, Text
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1] / "src"))
from brick_ar.infrastructure.tables import ADDED, md # noqa: E402
SRC = pathlib.Path(__file__).resolve().parents[1] / "src" / "brick_ar"
HEAD = ("-- Tables of the ar brick (module `ar`, invoicing). Every table prefixed `ar_`, a natural primary key each (never SERIAL),\n"
"-- money NUMERIC(18,2), a rate NUMERIC(19,4), creation columns last so a later column is an append. Additive only.\n"
"-- Dated rows (ar_settings, ar_customer, ar_account_map) are only ever inserted. *_versions tables are the optimistic-\n"
"-- version claims (erp_platform.repository): a save claims version n+1 with a plain INSERT, a racer fails 23505 -> 409.\n")
RELEASES = {2: ("audit1", "the proposer's and drafter's sub beside their email (self-approval holds for either identity),\n-- the invoice-number claim (no two documents share a number) and the per-invoice credit claim (credits cannot\n-- together exceed the invoice)."),
4: ("v4b", "spec v4 slice B: the e-invoice send and its statuses, deferred revenue releases, the doubtful-debt provision and its rates,\n-- cost centres on lines and the customer's e-invoice channel and default cost centre."),
5: ("v6", "spec v6: the partner functions on an invoice (sold-to, ship-to, bill-to) and the pricing quantity and unit on a line."),
6: ("v7", "spec v7: the quantity in the pricing unit and whether the net is the caller's exact one (a line's pricing conditions)."),
3: ("v4a", "spec v4 slice A: payment requests, deposit deductions with claims, settlement discounts, statutory interest,\n-- intercompany (company_code, trading_partner) and the hand-over.")}
def sql_type(t) -> str:
if isinstance(t, Numeric):
return f"NUMERIC({t.precision},{t.scale})"
for cls, name in ((BigInteger, "BIGINT"), (Integer, "INTEGER"), (Boolean, "BOOLEAN"), (TIMESTAMP, "TIMESTAMP"), (Date, "DATE"),
(Text, "VARCHAR"), (String, "VARCHAR")):
if isinstance(t, cls):
return name
raise TypeError(t)
def _later(table: str, column: str) -> bool:
"""Added after the first release (tables.ADDED)?"""
return any(spec == "*" and table == name or spec != "*" and table == name and column in spec
for added in ADDED.values() for name, spec in added.items())
def create(t, first: bool = False) -> str:
pk = [c.name for c in t.primary_key.columns]
cols = [c for c in t.columns if not (first and _later(t.name, c.name))]
width = max(len(c.name) for c in cols)
rows = [f" {c.name.ljust(width)} {sql_type(c.type)}{' NOT NULL' if c.primary_key or not c.nullable else ''}," for c in cols]
return f"CREATE TABLE IF NOT EXISTS {t.name} (\n" + "\n".join(rows) + f"\n PRIMARY KEY ({', '.join(pk)})\n);\n"
def main() -> int:
tables = list(md.sorted_tables)
schema = HEAD + "\n" + "\n".join(create(t) for t in tables)
first = [t for t in tables if not any(a.get(t.name) == "*" or a.get(t.name.removesuffix("_versions")) == "*" for a in ADDED.values())]
up = "-- 00001_init: the brick's first schema (the same tables as schema.sql).\n\n" + \
"\n".join(create(t, first=True) for t in first)
down = "-- 00001_init down: drops every table the brick made. Destroys the brick's data: a rollback of a first install only.\n" + \
"".join(f"DROP TABLE IF EXISTS {t.name};\n" for t in reversed(first))
files = {SRC / "schema.sql": schema, SRC / "migrations" / "00001_init" / "up.sql": up, SRC / "migrations" / "00001_init" / "down.sql": down}
for release, added in ADDED.items():
name, why = RELEASES[release]
d = SRC / "migrations" / f"{release:05d}_{name}"
ups = [f"-- {d.name} (v0.1.{release - 1}): {why} "
"ADD COLUMN is DISRUPTIVE: migrate the test copy first, the live one at a quiet hour.\n"]
downs = [f"-- {d.name} down: drops what the release added.\n"]
for name, spec in added.items():
if spec == "*":
for t in (name, name + "_versions"):
if t in md.tables:
ups.append(create(md.tables[t]))
downs.append(f"DROP TABLE IF EXISTS {t};\n")
else:
for col in spec:
ups.append(f"ALTER TABLE {name} ADD COLUMN {col} {sql_type(md.tables[name].c[col].type)};\n")
downs.append(f"ALTER TABLE {name} DROP COLUMN {col};\n")
files[d / "up.sql"], files[d / "down.sql"] = "\n".join(ups), "".join(downs)
if "--check" in sys.argv:
bad = [str(p) for p, body in files.items() if not p.exists() or p.read_text() != body]
print("out of date: " + ", ".join(bad) if bad else "schema files match the model")
return 1 if bad else 0
for p, body in files.items():
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text(body)
print(f"wrote {len(tables)} tables")
return 0
if __name__ == "__main__":
sys.exit(main())