diff --git a/scripts/migrate_task_hierarchy.py b/scripts/migrate_task_hierarchy.py new file mode 100644 index 0000000..4c3be88 --- /dev/null +++ b/scripts/migrate_task_hierarchy.py @@ -0,0 +1,49 @@ +#!/usr/bin/env python3 +"""一次性迁移:pipeline_tasks 加 parent_id + depends_on 列(幂等,information_schema 查列存在性)。 + +用途:任务分解与编排特性——父子关系(parent_id) + 依赖串行(depends_on)。 +已在 create_tables.py 的 _V2_COLUMNS 固化(供 build.sh 全新部署),本脚本供存量环境单独跑迁移。 +""" +import sys +import asyncio + +ROOT = '/d/pipeline/pipeline-app' +sys.path.insert(0, ROOT + '/py3/lib/python3.10/site-packages') +sys.path.insert(0, ROOT) + +from sqlor.dbpools import DBPools +from appPublic.jsonConfig import getConfig +from appPublic.folderUtils import ProgramPath + +COLS = { + 'parent_id': 'varchar(32) NULL', + 'depends_on': 'text NULL', +} + + +async def main(): + config = getConfig(ROOT, NS={'workdir': ROOT, 'ProgramPath': ProgramPath()}) + DBPools(config.databases) + async with DBPools().sqlorContext('pipeline') as sor: + for col, ddl in COLS.items(): + r = await sor.sqlExe( + "SELECT COUNT(*) as c FROM information_schema.COLUMNS " + "WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='pipeline_tasks' " + "AND COLUMN_NAME=${col}$", {'col': col}) + await sor.sqlExe("COMMIT", {}) + if not r or getattr(r[0], 'c', 0) == 0: + await sor.execute("ALTER TABLE pipeline_tasks ADD COLUMN %s %s" % (col, ddl), {}) + print(' added column:', col) + else: + print(' exists column:', col) + # 校验 + chk = await sor.sqlExe( + "SELECT COLUMN_NAME FROM information_schema.COLUMNS " + "WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='pipeline_tasks' " + "AND COLUMN_NAME IN ('parent_id','depends_on')", {}) + await sor.sqlExe("COMMIT", {}) + print(' verify columns:', [getattr(x, 'COLUMN_NAME', '') for x in (chk or [])]) + print('migration done') + + +asyncio.run(main())