dodil/erp-brick-gl

DODIL ERP brick: general ledger

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_gl.infrastructure.tables import ADDED, md  # noqa: E402

SRC = pathlib.Path(__file__).resolve().parents[1] / "src" / "brick_gl"
HEAD = ("-- Tables of the gl brick (module `gl`, the books). Every table prefixed `gl_`, a natural primary key each (never SERIAL),\n"
        "-- money NUMERIC(18,2), an exchange rate NUMERIC(19,6), a statistical quantity NUMERIC(19,4), creation columns last so a\n"
        "-- later column is an append. Additive only. Journals, lines, split lines, rates and trails are only ever inserted;\n"
        "-- gl_number_series and gl_chain_claim are claims (a plain INSERT: two writers cannot both hold one key).\n")


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)


RELEASES = {2: ("v6", "spec v6: when a migration run was reconciled (GET /migration/status).")}


def _later(table: str, column: str) -> bool:
    return any(table == name and column in cols for added in ADDED.values() for name, cols 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)
    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 tables)
    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(tables))
    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}: {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 table, cols in added.items():
            for col in cols:
                ups.append(f"ALTER TABLE {table} ADD COLUMN {col} {sql_type(md.tables[table].c[col].type)};\n")
                downs.append(f"ALTER TABLE {table} 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())