ute2e/tools/workflow-migration.py

"""Verify and apply the one workflow status constraint; no business data is rewritten."""
import runpy, json, hashlib, datetime
from pathlib import Path
import sqlparse

support=runpy.run_path(str(Path(__file__).with_name('review-db.py')))
source=Path(__file__).resolve().parents[2]/'unreal_tran_api/database/20260912_040_teaching_workflow_steps.sql'
report=Path(__file__).resolve().parents[1]/'reports/virtual-html-full-closure-20260912/migration.json'
report.parent.mkdir(parents=True,exist_ok=True)
connection=support['connect']()
with connection.cursor() as cursor:
    cursor.execute("SELECT pg_get_constraintdef(oid) AS definition FROM pg_constraint WHERE conname='ck_ut_teaching_step_run_status' AND conrelid='public.ut_teaching_step_run'::regclass")
    before=cursor.fetchone()
    cursor.execute('SHOW server_encoding')
    encoding=cursor.fetchone()
    # Temporary shadow table verifies syntax, first execution and repeat without touching public data.
    cursor.execute('CREATE TEMP TABLE ut_teaching_step_run (status_code varchar(32))')
    for attempt in range(2):
        for statement in sqlparse.split(source.read_text(encoding='utf-8')):cursor.execute(statement)
    cursor.execute("INSERT INTO pg_temp.ut_teaching_step_run VALUES ('SKIPPED')")
    cursor.execute('DROP TABLE pg_temp.ut_teaching_step_run')
    report.write_text(json.dumps({'file':source.name,'sha256':hashlib.sha256(source.read_bytes()).hexdigest(),'target':'unreal_tran/public','encoding':encoding,'before':before,'temporaryFirstAndRepeat':True,'stage':'before-public-apply'},ensure_ascii=False,indent=2),encoding='utf-8')
    for attempt in range(2):
        for statement in sqlparse.split(source.read_text(encoding='utf-8')):cursor.execute(statement)
    cursor.execute("SELECT pg_get_constraintdef(oid) AS definition FROM pg_constraint WHERE conname='ck_ut_teaching_step_run_status' AND conrelid='public.ut_teaching_step_run'::regclass")
    after=cursor.fetchone();assert 'SKIPPED' in after['definition']
    data=json.loads(report.read_text(encoding='utf-8'));data.update(after=after,stage='verified',time=datetime.datetime.now().astimezone().isoformat())
    report.write_text(json.dumps(data,ensure_ascii=False,indent=2),encoding='utf-8')
connection.close()
print('PASS temporary migration, repeat, public migration and constraint verification')