rmancinasandClaude Opus 4.8
fa9b696752
144 customers owned zero properties, zero policies and zero transactions — the legacy DATGRAL row exists but nothing in either business line ever attached to it. They padded the staff customer list with rows that can't be acted on. 27 were also nameless (dead ID slots); the other 117 have real names and sometimes contact details, and read as never-activated prospects or lapsed clients rather than junk. Removing both sets is a deliberate call. Implemented as a separate step rather than a filter inside transform_customers.py: emptiness is only knowable after properties, policies and transactions have loaded, and deciding it there would mean re-deriving each downstream transform's source-matching logic against the staged Parquet. Runs after transform_transactions.py in run_all.py. Safe by construction — a customer with no rows in any of the three tables has nothing pointing at it, so the delete cannot orphan anything; only its own customer_legacy_refs go with it. The step asserts zero orphans afterwards. Every pruned customer is written to output/pruned_customers.csv with its legacy provenance before the delete, and --dry-run reports without touching anything. Nothing is unrecoverable: the Access sources are untouched and a pipeline run without this step brings them all back. Verified: full run_all.py pass ends at 1538 customers (from 1682), with 1519 properties / 2378 policies / 45861 transactions all intact and zero orphans. 17 nameless customers remain, all of which carry real records. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
77 lines
3.1 KiB
Python
77 lines
3.1 KiB
Python
"""
|
|
Run the full data migration against one environment, in dependency order.
|
|
|
|
Every step is idempotent (truncate + rebuild), so this is safe to re-run. The
|
|
target DB is chosen with --env (reads deploy/.env.<env>); the same staged
|
|
Parquet feeds every environment.
|
|
|
|
Prerequisites (once per environment, NOT done here):
|
|
1. DB stack deployed (deploy/jorgecuadros-db.stack.yml) and deploy/.env.<env> written.
|
|
2. Prisma schema pushed to it:
|
|
DATABASE_URL="<that env's url>" \
|
|
npx prisma@5 db push --schema=packages/database/prisma/schema.prisma
|
|
|
|
Then:
|
|
./.venv/bin/python run_all.py --env dev # data only (staging already present)
|
|
./.venv/bin/python run_all.py --env prod --stage # re-extract from Access first, then load
|
|
|
|
Reproducing dev -> prod is exactly `--env prod` (plus --stage if the staged
|
|
Parquet isn't present on the machine running it).
|
|
|
|
Note: the blob_extract step re-reads the original Access files directly (the
|
|
blobs are not in the staged Parquet), so the machine running this needs
|
|
SOURCE_ROOT + mdbtools + MinIO credentials even without --stage.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
HERE = Path(__file__).parent
|
|
PY = sys.executable # the venv python running this orchestrator
|
|
|
|
# Dependency order. Every step truncates what it owns, so anything downstream
|
|
# of a truncated table has to be rebuilt in the same pass — blob_extract is in
|
|
# this list because transform_properties and transform_policies truncate
|
|
# service_documents / policy_documents, which would otherwise leave the
|
|
# uploaded MinIO objects with no rows pointing at them.
|
|
STEPS = [
|
|
"transform_customers.py", # customers + customer_legacy_refs (everything FKs to these)
|
|
"transform_properties.py", # properties + services + trust accounts
|
|
"transform_policies.py", # policies + installments/vehicles/drivers/benef/claims/adjusters
|
|
"transform_transactions.py", # shared ledger + type_transactions + exchange_rates
|
|
"prune_empty_customers.py", # drop customers with no property/policy/transaction
|
|
"transform_bank.py", # SCOTHIA bank register (no customer FK; independent)
|
|
"blob_extract.py", # document pointers; must follow properties + policies
|
|
]
|
|
|
|
|
|
def run(cmd: list[str]) -> None:
|
|
print("+ " + " ".join(cmd), flush=True)
|
|
r = subprocess.run(cmd)
|
|
if r.returncode:
|
|
sys.exit(r.returncode)
|
|
|
|
|
|
def main() -> None:
|
|
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
ap.add_argument("--env", default="dev", help="target environment (reads deploy/.env.<env>)")
|
|
ap.add_argument("--stage", action="store_true",
|
|
help="re-run the raw staging load first (needs the Access files + mdbtools)")
|
|
args = ap.parse_args()
|
|
|
|
if args.stage:
|
|
run([PY, str(HERE / "load_staging.py"), "--output-dir", str(HERE / "output")])
|
|
|
|
for step in STEPS:
|
|
run([PY, str(HERE / step), "--env", args.env])
|
|
|
|
print(f"\n✓ migration complete for env={args.env}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|