001d0abca2
- 重构用户和角色模型，优化字段定义和关系 - 增强初始化数据脚本，支持数据更新检查 - 改进用户和角色API端点，增加验证逻辑 - 扩展Pydantic模型，分离请求和响应模式 - 自定义Swagger UI界面并优化API文档 - 移除测试文件并更新依赖项配置
111 lines
3.3 KiB
Python
111 lines
3.3 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
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
|
|
|
|
router = APIRouter(prefix="/roles", tags=["roles"])
|
|
|
|
# 创建FastCRUD实例
|
|
role_crud = FastCRUD(Role, RoleResponse)
|
|
|
|
|
|
@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="角色名已存在")
|
|
|
|
# 准备创建数据
|
|
create_data = role_data.model_dump()
|
|
|
|
# 使用FastCRUD创建角色
|
|
created_role = await role_crud.create(db, create_data)
|
|
return created_role
|
|
|
|
|
|
@router.get("/", response_model=list[RoleResponse])
|
|
async def get_roles(
|
|
db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user)
|
|
):
|
|
"""获取所有角色"""
|
|
result = await role_crud.get_multi(db)
|
|
return result["data"]
|
|
|
|
|
|
@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),
|
|
):
|
|
"""获取单个角色"""
|
|
role = await role_crud.get(db, {"id": role_id})
|
|
if not role:
|
|
raise HTTPException(status_code=404, detail="角色不存在")
|
|
return 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="角色名已存在")
|
|
|
|
# 准备更新数据
|
|
update_data = role_data.model_dump()
|
|
|
|
# 使用FastCRUD更新角色
|
|
updated_role = await role_crud.update(db, {"id": role_id}, update_data)
|
|
if not updated_role:
|
|
raise HTTPException(status_code=404, detail="角色不存在")
|
|
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)} 个用户使用，无法删除"
|
|
)
|
|
|
|
# 使用FastCRUD删除角色
|
|
deleted = await role_crud.delete(db, {"id": role_id})
|
|
if not deleted:
|
|
raise HTTPException(status_code=404, detail="角色不存在")
|
|
return {"message": "角色删除成功"}
|