dodil/erp-brick-inventory

DODIL ERP brick: inventory

scripts/gen_schema.py

"""Write schema.sql and migrations/00001_init/up.sql from the column model in infrastructure/tables.py.
Run: python scripts/gen_schema.py   (tests/test_schema.py fails when the files and the model drift)."""
import pathlib
import sys

from sqlalchemy import TIMESTAMP as TS
from sqlalchemy import BigInteger, Boolean, Date, Integer, Numeric, String

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


def sql_type(t) -> str:
    if isinstance(t, Numeric):
        return f"NUMERIC({t.precision},{t.scale})"
    return {String: "VARCHAR", Integer: "INTEGER", BigInteger: "BIGINT", Boolean: "BOOLEAN", Date: "DATE", TS: "TIMESTAMP"}[type(t)]


def render() -> str:
    out = []
    for t in md.sorted_tables:
        cols = [f"    {c.name:<24}{sql_type(c.type)}{' NOT NULL' if c.primary_key else ''}" for c in t.columns]
        pks = ", ".join(c.name for c in t.primary_key.columns)
        out.append(f"CREATE TABLE IF NOT EXISTS {t.name} (\n" + ",\n".join(cols) + f",\n    PRIMARY KEY ({pks})\n);")
    return "\n".join(out) + "\n"


HEAD = ("-- Tables of the inventory brick (module `inventory`, spec v3). Every table prefixed `inv_`, a natural primary key each\n"
        "-- (never SERIAL), quantities NUMERIC(19,4) in the product's base unit, money NUMERIC(18,2), costs NUMERIC(19,4).\n"
        "-- Generated from infrastructure/tables.py by scripts/gen_schema.py. Additive only: a new column goes at the END.\n\n")

if __name__ == "__main__":
    body = HEAD + render()
    (ROOT / "src/brick_inventory/schema.sql").write_text(body)
    mig = ROOT / "src/brick_inventory/migrations/00001_init"
    mig.mkdir(parents=True, exist_ok=True)
    (mig / "up.sql").write_text("-- 00001_init: the brick's first schema (same tables as schema.sql).\n" + body)
    (mig / "down.sql").write_text("-- 00001_init down: drops every table the brick made. Destroys the brick's data: a rollback of a first install only.\n"
                                  + "\n".join(f"DROP TABLE IF EXISTS {t.name};" for t in reversed(md.sorted_tables)) + "\n")
    print(f"{len(md.sorted_tables)} tables")