rmancinasandClaude Opus 4.8
1b79b43a54
The --sync path had never been run and was broken in several ways. Fixed and verified against the dev DB (two consecutive syncs, both exit 0, 32/32 assertions: stable PKs, manual-row preservation, changed-row updates, legacy-delete, no child duplication, zero FK orphans; idempotent). - policies/properties: reuse each legacy row's existing id (by provenance) BEFORE building child rows, so children no longer point at a discarded fresh uuid; rebuild legacy-owned children via scoped delete + reinsert. - customers: replace zip(customers, refs) (mispaired almost every row) with a ref-grouped id remap; names now restore and no spurious customers appear. - drop the invalid Vehicle @@unique(legacySourceTable, legacyId) — one legacy policy row carries up to 3 vehicles sharing a legacyId; handle via delete+reinsert. - upsert lookup tables (policy_types, insurance_providers, type_transactions, adjusters) by natural name and remap child FKs instead of inserting fresh uuids that nothing points at. - transactions: drop updatedAt=NOW() (no such column); guard report formatting on NULL legacySourceTable (manual rows). Same report guard in bank. - add manual-safe prune (prune_empty_customers.py --sync, in SYNC_STEPS): prune only legacy-owned empties, never manually-added customers. web: customer-detail mini tx list now strikes voided rows with an "(anulado)" tag (was the last void-UI rendering gap; /estado-cuenta already handled it). docs: RESUME.md updated — Phase B sync marked verified end-to-end, void-UI browser pass recorded. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
93 lines
3.2 KiB
Python
93 lines
3.2 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",
|
|
"transform_properties.py",
|
|
"transform_policies.py",
|
|
"transform_transactions.py",
|
|
"prune_empty_customers.py",
|
|
"transform_bank.py",
|
|
"blob_extract.py",
|
|
]
|
|
|
|
SYNC_STEPS = [
|
|
"transform_customers.py",
|
|
"transform_properties.py",
|
|
"transform_policies.py",
|
|
"transform_transactions.py",
|
|
# Manual-safe prune: drops legacy-owned empties that the customer upsert
|
|
# re-creates from Parquet, but leaves manually-added customers alone.
|
|
"prune_empty_customers.py",
|
|
"transform_bank.py",
|
|
]
|
|
|
|
|
|
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)")
|
|
ap.add_argument("--sync", action="store_true",
|
|
help="upsert legacy rows and archive removed legacy rows; preserve manual rows")
|
|
args = ap.parse_args()
|
|
|
|
if args.stage:
|
|
run([PY, str(HERE / "load_staging.py"), "--output-dir", str(HERE / "output")])
|
|
|
|
for step in SYNC_STEPS if args.sync else STEPS:
|
|
cmd = [PY, str(HERE / step), "--env", args.env]
|
|
if args.sync:
|
|
cmd.append("--sync")
|
|
run(cmd)
|
|
|
|
print(f"\n✓ migration complete for env={args.env}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|