fix(ops): re-stage before an additive sync
SYNC ran `run_all.py --sync` without `--stage`, so it depended on staged
Parquet under migration/output. That directory is part of the image, not a
volume, so any redeploy wiped it and the job died on the first transform:
FileNotFoundError: '/repo/migration/output/stg_utilities/datgral.parquet'
Re-staging is also what makes the job's own label true — without it a sync
would replay whatever upload staged last, not the files currently in the
ingest folder.
Staging now counts as a numbered step when it runs, so the Operaciones
progress bar moves during the slowest phase instead of sitting empty.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
+17
-5
@@ -15,6 +15,11 @@ 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
|
||||
|
||||
--sync swaps the truncate+rebuild steps for the additive upsert ones. It reads
|
||||
the same staged Parquet, so it needs --stage too unless a previous run left
|
||||
migration/output populated on this machine — which is never true in a
|
||||
container, where that directory is part of the image and dies with it.
|
||||
|
||||
Reproducing dev -> prod is exactly `--env prod` (plus --stage if the staged
|
||||
Parquet isn't present on the machine running it).
|
||||
|
||||
@@ -100,15 +105,22 @@ def main() -> None:
|
||||
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")])
|
||||
|
||||
steps = SYNC_STEPS if args.sync else STEPS
|
||||
for i, step in enumerate(steps, start=1):
|
||||
# Staging counts as a step when it runs: it is the slowest part of the pass
|
||||
# (mdbtools re-reads every Access file), so leaving it outside the numbering
|
||||
# would park the Operaciones progress bar at "nothing yet" for minutes.
|
||||
total = len(steps) + (1 if args.stage else 0)
|
||||
offset = 1 if args.stage else 0
|
||||
|
||||
if args.stage:
|
||||
run([PY, str(HERE / "load_staging.py"), "--output-dir", str(HERE / "output")],
|
||||
step=1, total=total)
|
||||
|
||||
for i, step in enumerate(steps, start=1 + offset):
|
||||
cmd = [PY, str(HERE / step), "--env", args.env]
|
||||
if args.sync:
|
||||
cmd.append("--sync")
|
||||
run(cmd, step=i, total=len(steps))
|
||||
run(cmd, step=i, total=total)
|
||||
|
||||
print(f"\n✓ migration complete for env={args.env}")
|
||||
|
||||
|
||||
Reference in New Issue
Block a user