998211c483
- 从服务器拉取完整代码 - 按框架规范整理项目结构 - 配置 Drone CI 测试环境部署 - 包含后端(FastAPI)、前端(Vue3)、管理端 技术栈: Vue3 + TypeScript + FastAPI + MySQL
98 lines
3.1 KiB
Python
98 lines
3.1 KiB
Python
#!/usr/bin/env python
|
|
"""
|
|
调试用户更新API
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
# 添加项目根目录到 Python 路径
|
|
project_root = Path(__file__).parent
|
|
sys.path.append(str(project_root))
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from app.core.deps import get_db
|
|
from app.services.user_service import UserService
|
|
from app.schemas.user import UserUpdate
|
|
from app.models.user import User
|
|
from app.core.logger import logger
|
|
|
|
async def debug_update():
|
|
"""调试更新流程"""
|
|
|
|
# 获取数据库会话
|
|
async for db in get_db():
|
|
try:
|
|
# 1. 创建服务实例
|
|
user_service = UserService(db)
|
|
|
|
# 2. 获取superadmin用户
|
|
print("1. 获取superadmin用户...")
|
|
user = await user_service.get_by_username("superadmin")
|
|
if not user:
|
|
print("用户不存在")
|
|
return
|
|
|
|
print(f"用户ID: {user.id}")
|
|
print(f"用户名: {user.username}")
|
|
print(f"当前学校: {user.school}")
|
|
print(f"当前专业: {user.major}")
|
|
|
|
# 3. 创建更新数据
|
|
print("\n2. 创建更新数据...")
|
|
update_data = UserUpdate(
|
|
school="北京大学",
|
|
major="软件工程",
|
|
bio="测试更新"
|
|
)
|
|
|
|
print(f"更新数据: {update_data.model_dump(exclude_unset=True)}")
|
|
|
|
# 4. 执行更新
|
|
print("\n3. 执行更新...")
|
|
updated_user = await user_service.update_user(
|
|
user_id=user.id,
|
|
obj_in=update_data,
|
|
updated_by=user.id
|
|
)
|
|
|
|
# 5. 检查更新结果
|
|
print("\n4. 更新后的数据:")
|
|
print(f"用户ID: {updated_user.id}")
|
|
print(f"用户名: {updated_user.username}")
|
|
print(f"学校: {updated_user.school}")
|
|
print(f"专业: {updated_user.major}")
|
|
print(f"个人简介: {updated_user.bio}")
|
|
|
|
# 6. 再次查询验证
|
|
print("\n5. 再次查询验证...")
|
|
verified_user = await user_service.get_by_id(user.id)
|
|
print(f"验证 - 学校: {verified_user.school}")
|
|
print(f"验证 - 专业: {verified_user.major}")
|
|
|
|
# 7. 检查数据库中的实际值
|
|
print("\n6. 检查数据库实际值...")
|
|
from sqlalchemy import text
|
|
result = await db.execute(
|
|
text("SELECT school, major FROM users WHERE id = :user_id"),
|
|
{"user_id": user.id}
|
|
)
|
|
row = result.fetchone()
|
|
if row:
|
|
print(f"数据库 - 学校: {row[0]}")
|
|
print(f"数据库 - 专业: {row[1]}")
|
|
|
|
except Exception as e:
|
|
print(f"错误: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
finally:
|
|
await db.close()
|
|
break
|
|
|
|
if __name__ == "__main__":
|
|
print("调试用户更新API...")
|
|
asyncio.run(debug_update())
|