f1da855793
- 添加用户登录、登出及令牌管理功能 - 实现基于JWT和Redis的认证系统 - 完善用户和角色管理API - 添加密码加密与验证功能 - 配置数据库连接和Redis客户端 - 实现中间件进行权限验证 - 初始化管理员和测试用户数据 - 更新模型和接口文档 - 添加测试API和配置文件
103 lines
2.9 KiB
Python
103 lines
2.9 KiB
Python
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select
|
|
from models import Role, User
|
|
from utils.password import get_password_hash
|
|
|
|
|
|
async def init_roles(db: AsyncSession):
|
|
"""初始化角色数据"""
|
|
# 检查是否已经存在角色数据
|
|
result = await db.execute(select(Role))
|
|
existing_roles = result.scalars().all()
|
|
if existing_roles:
|
|
print("角色数据已存在，跳过初始化")
|
|
return
|
|
|
|
# 创建初始角色
|
|
roles = [
|
|
Role(name="管理员", creator="system"),
|
|
Role(name="普通用户", creator="system"),
|
|
Role(name="访客", creator="system"),
|
|
]
|
|
|
|
for role in roles:
|
|
db.add(role)
|
|
|
|
await db.commit()
|
|
print("角色数据初始化完成")
|
|
|
|
|
|
async def init_admin_user(db: AsyncSession):
|
|
"""初始化管理员用户"""
|
|
# 检查是否已经存在管理员用户
|
|
result = await db.execute(select(User).filter(User.username == "admin"))
|
|
existing_admin = result.scalar_one_or_none()
|
|
if existing_admin:
|
|
print("管理员用户已存在，跳过初始化")
|
|
return
|
|
|
|
# 获取管理员角色
|
|
result = await db.execute(select(Role).filter(Role.name == "管理员"))
|
|
admin_role = result.scalar_one_or_none()
|
|
if not admin_role:
|
|
print("管理员角色不存在，请先初始化角色数据")
|
|
return
|
|
|
|
# 创建管理员用户
|
|
admin_user = User(
|
|
username="admin",
|
|
nickname="系统管理员",
|
|
email="admin@example.com",
|
|
phone="13800138000",
|
|
password_hash=get_password_hash("admin"),
|
|
role_id=admin_role.id,
|
|
avatar="",
|
|
wx_openid=""
|
|
)
|
|
|
|
db.add(admin_user)
|
|
await db.commit()
|
|
print("管理员用户初始化完成")
|
|
|
|
|
|
async def init_test_user(db: AsyncSession):
|
|
"""初始化测试用户"""
|
|
# 检查是否已经存在测试用户
|
|
result = await db.execute(select(User).filter(User.username == "test"))
|
|
existing_test_user = result.scalar_one_or_none()
|
|
if existing_test_user:
|
|
print("测试用户已存在，跳过初始化")
|
|
return
|
|
|
|
# 获取普通用户角色
|
|
result = await db.execute(select(Role).filter(Role.name == "普通用户"))
|
|
user_role = result.scalar_one_or_none()
|
|
if not user_role:
|
|
print("普通用户角色不存在，请先初始化角色数据")
|
|
return
|
|
|
|
# 创建测试用户
|
|
test_user = User(
|
|
username="test",
|
|
nickname="测试用户",
|
|
email="test@example.com",
|
|
phone="13800138001",
|
|
password_hash=get_password_hash("test"),
|
|
role_id=user_role.id,
|
|
avatar="",
|
|
wx_openid=""
|
|
)
|
|
|
|
db.add(test_user)
|
|
await db.commit()
|
|
print("测试用户初始化完成")
|
|
|
|
|
|
async def init_all_data(db: AsyncSession):
|
|
"""初始化所有数据"""
|
|
print("开始初始化数据库数据...")
|
|
await init_roles(db)
|
|
await init_admin_user(db)
|
|
await init_test_user(db)
|
|
print("数据库数据初始化完成")
|