File size: 2,047 Bytes
ef983af | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 | #!/usr/bin/env python3
"""Initialize (or upgrade) the D3 financial schema on an explicit local database — idempotent.
Creates the D3 tables on the shared D2 SQLite database and ensures a tenant's system ledger
accounts. Never seeds balances or merchants merely because it runs. Requires --confirm-d3.
Example:
python scripts/d3_initialize_finance.py --db-path /tmp/amanpay-d2/amanpay.db --tenant event \
--confirm-d3 --allow-unknown-fs --json
"""
from __future__ import annotations
import argparse
import sys
from amanpay.finance.chart_of_accounts import ensure_system_accounts
from amanpay.finance.cli import (add_common_args, build_runtime_for_cli, emit, resolve_tenant_id)
from amanpay.finance.storage.store import d3_schema_version
def main(argv=None) -> int:
p = argparse.ArgumentParser(description="Initialize the D3 financial schema (idempotent).")
add_common_args(p, tenant_required=False)
p.add_argument("--confirm-d3", action="store_true", required=True,
help="explicit confirmation to create/upgrade D3 tables")
args = p.parse_args(argv)
rt = build_runtime_for_cli(args.db_path, allow_unknown_fs=args.allow_unknown_fs)
out = {"db_path": args.db_path, "d3_schema_version": d3_schema_version(rt.store),
"journal_mode": rt.store.runtime_report().get("journal_mode")}
if args.tenant:
tid = resolve_tenant_id(rt, args.tenant)
if tid is None:
# Schema init (the primary job) still succeeded; system accounts need an existing
# tenant (created via the D2 CLIs / seed). Report and continue — not a hard error.
out["tenant_status"] = "not_found"
out["system_accounts"] = 0
else:
created = ensure_system_accounts(rt.repo, tid)
out["tenant_id"] = tid
out["tenant_status"] = "ok"
out["system_accounts"] = len(created)
out["ok"] = True
emit(out, as_json=args.json)
rt.close()
return 0
if __name__ == "__main__":
sys.exit(main())
|