84 lines
2.8 KiB
Python
84 lines
2.8 KiB
Python
#!/usr/bin/env python3
|
|
"""应用测试单元表字段迁移"""
|
|
|
|
import asyncio
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
from sqlalchemy import text
|
|
|
|
DB_CONFIG = {
|
|
'host': '123.57.81.127',
|
|
'port': 3306,
|
|
'user': 'cpy_admin',
|
|
'password': 'Tgzz2025+',
|
|
'database': 'ruoyi-fastapi'
|
|
}
|
|
|
|
DATABASE_URL = (
|
|
f"mysql+asyncmy://{DB_CONFIG['user']}:{DB_CONFIG['password']}@"
|
|
f"{DB_CONFIG['host']}:{DB_CONFIG['port']}/{DB_CONFIG['database']}"
|
|
)
|
|
|
|
async def apply_migration():
|
|
engine = create_async_engine(DATABASE_URL, echo=False)
|
|
|
|
async with engine.connect() as conn:
|
|
print("=" * 80)
|
|
print("Applying Test Item Fields Migration")
|
|
print("=" * 80)
|
|
|
|
# 1. 检查现有字段
|
|
print("\n[1] Checking current test_item structure...")
|
|
result = await conn.execute(text("DESCRIBE test_item"))
|
|
columns = result.fetchall()
|
|
column_names = [col[0] for col in columns]
|
|
print(f" Existing columns: {', '.join(column_names)}")
|
|
|
|
# 2. 添加 condition_json 字段
|
|
if 'condition_json' not in column_names:
|
|
print("\n[2] Adding condition_json column...")
|
|
await conn.execute(text("""
|
|
ALTER TABLE test_item
|
|
ADD COLUMN condition_json TEXT NULL COMMENT '测试条件模板JSON' AFTER eut_type_id
|
|
"""))
|
|
await conn.commit()
|
|
print(" [OK] condition_json added")
|
|
else:
|
|
print("\n[2] condition_json already exists")
|
|
|
|
# 3. 修改 config_json 字段为 TEXT 类型
|
|
print("\n[3] Modifying config_json column...")
|
|
await conn.execute(text("""
|
|
ALTER TABLE test_item
|
|
MODIFY COLUMN config_json TEXT NULL COMMENT '测试结果模板JSON'
|
|
"""))
|
|
await conn.commit()
|
|
print(" [OK] config_json modified")
|
|
|
|
# 4. 添加 raw_data_fields 字段
|
|
if 'raw_data_fields' not in column_names:
|
|
print("\n[4] Adding raw_data_fields column...")
|
|
await conn.execute(text("""
|
|
ALTER TABLE test_item
|
|
ADD COLUMN raw_data_fields TEXT NULL COMMENT '原始数据字段JSON' AFTER config_json
|
|
"""))
|
|
await conn.commit()
|
|
print(" [OK] raw_data_fields added")
|
|
else:
|
|
print("\n[4] raw_data_fields already exists")
|
|
|
|
# 5. 验证结果
|
|
print("\n[5] Verification:")
|
|
result = await conn.execute(text("DESCRIBE test_item"))
|
|
columns = result.fetchall()
|
|
for col in columns:
|
|
print(f" {col[0]:<20} | {col[1]:<30} | {col[2]}")
|
|
|
|
print("\n" + "=" * 80)
|
|
print("Migration Complete")
|
|
print("=" * 80)
|
|
|
|
await engine.dispose()
|
|
|
|
if __name__ == '__main__':
|
|
asyncio.run(apply_migration())
|