2641352f5e
1. 为角色、权限、用户的增删改请求统一添加强制成功提示 2. 移除onSuccess回调中的重复成功提示逻辑 3. 关闭表格删除操作的内置成功通知，避免重复弹窗 4. 重构用户管理页面，移除手动绑定的成功/错误回调，交由crud统一处理 5. 修复后端角色模块的FastCRUD使用问题，改用原生SQLAlchemy操作数据库
149 lines
4.4 KiB
Python
149 lines
4.4 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from sqlalchemy import select
|
||
from fastcrud import FastCRUD
|
||
|
||
from database import get_db
|
||
from models import Role, User
|
||
from schemas import RoleCreate, RoleResponse
|
||
from middleware import get_current_user
|
||
from utils.like_filter import ilike_contains
|
||
|
||
router = APIRouter(prefix="/roles", tags=["roles"])
|
||
|
||
# 创建FastCRUD实例，指定update_schema为RoleCreate
|
||
role_crud = FastCRUD(Role, RoleResponse, RoleCreate)
|
||
|
||
|
||
@router.post("/", response_model=RoleResponse)
|
||
async def create_role(
|
||
role_data: RoleCreate,
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""创建角色"""
|
||
# 检查角色名是否已存在
|
||
from sqlalchemy import select
|
||
|
||
result = await db.execute(select(Role).filter(Role.name == role_data.name))
|
||
existing_role = result.scalar_one_or_none()
|
||
if existing_role:
|
||
raise HTTPException(status_code=400, detail="角色名已存在")
|
||
|
||
# 设置创建人为当前登录用户
|
||
role_data.creator = current_user.nickname or current_user.username
|
||
|
||
# 创建角色对象
|
||
new_role = Role(
|
||
name=role_data.name,
|
||
creator=role_data.creator
|
||
)
|
||
|
||
db.add(new_role)
|
||
await db.flush() # 获取生成的ID
|
||
await db.refresh(new_role)
|
||
await db.commit()
|
||
|
||
return RoleResponse.model_validate(new_role)
|
||
|
||
|
||
@router.post("/search", response_model=list[RoleResponse])
|
||
async def search_roles(
|
||
query: dict = {},
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""获取角色列表（POST方式）；支持按名称模糊查询"""
|
||
stmt = select(Role)
|
||
name = query.get("name")
|
||
cond = ilike_contains(Role.name, name)
|
||
if cond is not None:
|
||
stmt = stmt.where(cond)
|
||
result = await db.execute(stmt)
|
||
rows = result.scalars().all()
|
||
return [RoleResponse.model_validate(r) for r in rows]
|
||
|
||
|
||
@router.get("/{role_id}", response_model=RoleResponse)
|
||
async def get_role(
|
||
role_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""获取单个角色"""
|
||
result = await db.execute(select(Role).filter(Role.id == role_id))
|
||
role = result.scalar_one_or_none()
|
||
if not role:
|
||
raise HTTPException(status_code=404, detail="角色不存在")
|
||
return RoleResponse.model_validate(role)
|
||
|
||
|
||
@router.put("/{role_id}", response_model=RoleResponse)
|
||
async def update_role(
|
||
role_id: int,
|
||
role_data: RoleCreate,
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""更新角色"""
|
||
# 检查角色名是否被其他角色使用
|
||
from sqlalchemy import select
|
||
|
||
result = await db.execute(
|
||
select(Role).filter(Role.name == role_data.name, Role.id != role_id)
|
||
)
|
||
existing_role = result.scalar_one_or_none()
|
||
if existing_role:
|
||
raise HTTPException(status_code=400, detail="角色名已存在")
|
||
|
||
# 使用原生 SQLAlchemy 更新角色（FastCRUD 有问题）
|
||
from sqlalchemy import update
|
||
|
||
stmt = (
|
||
update(Role)
|
||
.where(Role.id == role_id)
|
||
.values(name=role_data.name)
|
||
)
|
||
result = await db.execute(stmt)
|
||
|
||
if result.rowcount == 0:
|
||
raise HTTPException(status_code=404, detail="角色不存在")
|
||
|
||
await db.commit()
|
||
|
||
# 查询更新后的数据
|
||
result = await db.execute(select(Role).filter(Role.id == role_id))
|
||
updated_role = result.scalar_one_or_none()
|
||
|
||
return updated_role
|
||
|
||
|
||
@router.delete("/{role_id}")
|
||
async def delete_role(
|
||
role_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""删除角色"""
|
||
# 检查是否有用户使用此角色
|
||
from sqlalchemy import select
|
||
|
||
result = await db.execute(select(User).filter(User.role_id == role_id))
|
||
users = result.scalars().all()
|
||
if users:
|
||
raise HTTPException(
|
||
status_code=400, detail=f"该角色正在被 {len(users)} 个用户使用，无法删除"
|
||
)
|
||
|
||
# 使用原生 SQLAlchemy 删除角色（FastCRUD 有问题）
|
||
from sqlalchemy import delete as sql_delete
|
||
|
||
stmt = sql_delete(Role).where(Role.id == role_id)
|
||
result = await db.execute(stmt)
|
||
|
||
if result.rowcount == 0:
|
||
raise HTTPException(status_code=404, detail="角色不存在")
|
||
|
||
await db.commit()
|
||
return {"message": "角色删除成功"}
|