64 lines
2.4 KiB
Python
64 lines
2.4 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 create_table():
|
||
|
|
engine = create_async_engine(DATABASE_URL, echo=False)
|
||
|
|
|
||
|
|
async with engine.connect() as conn:
|
||
|
|
print("=" * 80)
|
||
|
|
print("Creating test_form table")
|
||
|
|
print("=" * 80)
|
||
|
|
|
||
|
|
# 创建测试表单表
|
||
|
|
await conn.execute(text("""
|
||
|
|
CREATE TABLE IF NOT EXISTS test_form (
|
||
|
|
id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '表单ID',
|
||
|
|
name VARCHAR(100) NOT NULL COMMENT '表单名称',
|
||
|
|
form_type VARCHAR(20) NOT NULL COMMENT '表单类型: CONDITION=测试条件, RESULT=测试结果',
|
||
|
|
form_json TEXT NOT NULL COMMENT '表单JSON定义',
|
||
|
|
description VARCHAR(500) COMMENT '表单说明',
|
||
|
|
is_active TINYINT(1) DEFAULT 1 COMMENT '是否启用: 0=禁用, 1=启用',
|
||
|
|
del_flag VARCHAR(1) DEFAULT '0' COMMENT '删除标志: 0=正常, 1=已删除',
|
||
|
|
create_by VARCHAR(64) COMMENT '创建者',
|
||
|
|
create_time DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
|
||
|
|
update_by VARCHAR(64) COMMENT '更新者',
|
||
|
|
update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
|
||
|
|
remark VARCHAR(500) COMMENT '备注',
|
||
|
|
INDEX idx_form_type (form_type),
|
||
|
|
INDEX idx_is_active (is_active)
|
||
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='测试表单模板表'
|
||
|
|
"""))
|
||
|
|
await conn.commit()
|
||
|
|
print(" [OK] test_form table created")
|
||
|
|
|
||
|
|
# 验证表结构
|
||
|
|
print("\n[Verification]")
|
||
|
|
result = await conn.execute(text("DESCRIBE test_form"))
|
||
|
|
columns = result.fetchall()
|
||
|
|
for col in columns:
|
||
|
|
print(f" {col[0]:<20} | {col[1]:<30}")
|
||
|
|
|
||
|
|
print("\n" + "=" * 80)
|
||
|
|
|
||
|
|
await engine.dispose()
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
asyncio.run(create_table())
|