998211c483
- 从服务器拉取完整代码 - 按框架规范整理项目结构 - 配置 Drone CI 测试环境部署 - 包含后端(FastAPI)、前端(Vue3)、管理端 技术栈: Vue3 + TypeScript + FastAPI + MySQL
95 lines
3.2 KiB
Python
95 lines
3.2 KiB
Python
"""
|
||
同步 users 表缺失列（适配当前模型定义）。
|
||
|
||
- 逐列检查 INFORMATION_SCHEMA，缺失则执行 ALTER TABLE 添加
|
||
- 避免一次性重建表，降低风险
|
||
|
||
运行方式：
|
||
cd kaopeilian-backend && python3 scripts/sync_users_table.py
|
||
"""
|
||
import asyncio
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
# 确保可导入应用配置
|
||
sys.path.append(str(Path(__file__).resolve().parent.parent))
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.ext.asyncio import create_async_engine
|
||
|
||
from app.core.config import get_settings
|
||
|
||
|
||
async def column_exists(conn, db_name: str, table: str, column: str) -> bool:
|
||
"""检查列是否存在"""
|
||
result = await conn.execute(
|
||
text(
|
||
"""
|
||
SELECT COUNT(1)
|
||
FROM INFORMATION_SCHEMA.COLUMNS
|
||
WHERE TABLE_SCHEMA = :db
|
||
AND TABLE_NAME = :table
|
||
AND COLUMN_NAME = :column
|
||
"""
|
||
),
|
||
{"db": db_name, "table": table, "column": column},
|
||
)
|
||
return (result.scalar() or 0) > 0
|
||
|
||
|
||
async def sync_users_table():
|
||
"""对 users 表进行列同步（仅添加缺失列）。"""
|
||
settings = get_settings()
|
||
|
||
# 解析数据库名
|
||
db_url = settings.DATABASE_URL
|
||
db_name = db_url.split("/")[-1].split("?")[0]
|
||
|
||
engine = create_async_engine(settings.DATABASE_URL, echo=False)
|
||
|
||
async with engine.begin() as conn:
|
||
# 需要确保存在的列（列名 -> DDL 片段）
|
||
required_columns = {
|
||
"avatar_url": "ALTER TABLE `users` ADD COLUMN `avatar_url` VARCHAR(500) NULL COMMENT '头像URL'",
|
||
"bio": "ALTER TABLE `users` ADD COLUMN `bio` TEXT NULL COMMENT '个人简介'",
|
||
"role": "ALTER TABLE `users` ADD COLUMN `role` VARCHAR(20) NOT NULL DEFAULT 'trainee' COMMENT '系统角色'",
|
||
"is_active": "ALTER TABLE `users` ADD COLUMN `is_active` BOOLEAN NOT NULL DEFAULT TRUE COMMENT '是否激活'",
|
||
"is_verified": "ALTER TABLE `users` ADD COLUMN `is_verified` BOOLEAN NOT NULL DEFAULT FALSE COMMENT '是否验证'",
|
||
"last_login_at": "ALTER TABLE `users` ADD COLUMN `last_login_at` DATETIME NULL COMMENT '最后登录时间'",
|
||
"password_changed_at": "ALTER TABLE `users` ADD COLUMN `password_changed_at` DATETIME NULL COMMENT '密码修改时间'",
|
||
"is_deleted": "ALTER TABLE `users` ADD COLUMN `is_deleted` BOOLEAN NOT NULL DEFAULT FALSE COMMENT '是否删除'",
|
||
"deleted_at": "ALTER TABLE `users` ADD COLUMN `deleted_at` DATETIME NULL COMMENT '删除时间'",
|
||
}
|
||
|
||
applied = []
|
||
for col, ddl in required_columns.items():
|
||
exists = await column_exists(conn, db_name, "users", col)
|
||
if not exists:
|
||
await conn.execute(text(ddl))
|
||
applied.append(col)
|
||
|
||
await engine.dispose()
|
||
return applied
|
||
|
||
|
||
async def main():
|
||
try:
|
||
applied = await sync_users_table()
|
||
if applied:
|
||
print("添加列:", ", ".join(applied))
|
||
else:
|
||
print("users 表结构已满足要求，无需变更。")
|
||
except Exception as exc:
|
||
# 输出完整错误，便于调试
|
||
import traceback
|
||
|
||
print("同步失败:", str(exc))
|
||
traceback.print_exc()
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|
||
|
||
|