dodil/erp-brick-pos

DODIL ERP brick: point of sale

scripts/gen_schema.py

"""Write schema.sql and migrations/NNNNN_<name>/{up,down}.sql from the column model in infrastructure/tables.py.

    python scripts/gen_schema.py            # rewrite the 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 and tables (never drop, rename or retype): the model lists them in
`ADDED` (columns) and `NEW_TABLES` (tables) by release, 00001_init is written without them and each release's migration has them.
"""
import pathlib
import sys

from sqlalchemy import TIMESTAMP, Boolean, Date, Integer, Numeric, String

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1] / "src"))
from brick_pos.infrastructure.tables import ADDED, NEW_TABLES, md  # noqa: E402

SRC = pathlib.Path(__file__).resolve().parents[1] / "src" / "brick_pos"
SCHEMA_HEAD = ("-- pos v1, slice A: every table prefixed `pos_`, a natural primary key each (never SERIAL), money NUMERIC(18,2), a unit price\n"
               "-- NUMERIC(19,4). Additive only: a later slice appends tables and columns, never drops or renames.\n")
INIT_HEAD = ("-- 00001_init: the brick's first schema (pos v1, slice A). The versioned layout migrations/NNNNN_<name>/{up,down}.sql with a\n"
             "-- per-brick journal, pos_schema_migrations. A later slice adds 00002_<name>; it never edits this file.\n\n"
             "CREATE TABLE IF NOT EXISTS pos_schema_migrations (\n    migration  VARCHAR NOT NULL,\n    applied_at TIMESTAMP NOT NULL,\n"
             "    PRIMARY KEY (migration)\n);\n")
RELEASES = {2: ("slice_b", "slice B: closing the day, refunds, vouchers, recipes, the register snapshot.")}


def sql_type(t) -> str:
    if isinstance(t, Numeric):
        return f"NUMERIC({t.precision},{t.scale})"
    for cls, name in ((Integer, "INTEGER"), (Boolean, "BOOLEAN"), (TIMESTAMP, "TIMESTAMP"), (Date, "DATE"), (String, "VARCHAR")):
        if isinstance(t, cls):
            return name
    raise TypeError(t)


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 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)
    later = {n for names in NEW_TABLES.values() for n in names}
    first = [t for t in tables if t.name not in later]
    files = {SRC / "schema.sql": SCHEMA_HEAD + "\n" + "\n".join(create(t) for t in tables),
             SRC / "migrations" / "00001_init" / "up.sql": INIT_HEAD + "\n" + "\n".join(create(t, first=True) for t in first),
             SRC / "migrations" / "00001_init" / "down.sql":
                 "-- 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)) + "DROP TABLE IF EXISTS pos_schema_migrations;\n"}
    for release in sorted({*ADDED, *NEW_TABLES}):
        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 tname in NEW_TABLES.get(release, []):
            ups.append(create(md.tables[tname]))
        for table, cols in ADDED.get(release, {}).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")
        downs += [f"DROP TABLE IF EXISTS {n};\n" for n in reversed(NEW_TABLES.get(release, []))]
        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())