feat: migrate_task_hierarchy.py 存量环境幂等迁移 parent_id/depends_on(避开 create_tables.py 全量 DROP 业务表)
This commit is contained in:
parent
3d55d590bf
commit
16ab9f73a0
49
scripts/migrate_task_hierarchy.py
Normal file
49
scripts/migrate_task_hierarchy.py
Normal file
@ -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())
|
||||||
Loading…
x
Reference in New Issue
Block a user