100 lines
3.4 KiB
Python
100 lines
3.4 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
from app.core.db import get_db
|
|
from app.schemas.role import RoleOut, RoleCreate, RoleUpdate, RolePermissionUpdate
|
|
from app.models import Role, RolePermission, Permission, UserRole
|
|
from app.models.enums import StatusEnum
|
|
|
|
|
|
router = APIRouter(prefix="/roles", tags=["roles"])
|
|
|
|
|
|
@router.get("", response_model=list[RoleOut])
|
|
def list_roles(db: Session = Depends(get_db)):
|
|
return db.query(Role).all()
|
|
|
|
|
|
@router.post("", response_model=RoleOut)
|
|
def create_role(payload: RoleCreate, db: Session = Depends(get_db)):
|
|
exists = db.query(Role).filter(Role.role_code == payload.role_code).first()
|
|
if exists:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Role code exists")
|
|
role = Role(
|
|
role_code=payload.role_code,
|
|
role_name=payload.role_name,
|
|
status=payload.status,
|
|
remark=payload.remark,
|
|
)
|
|
db.add(role)
|
|
db.commit()
|
|
db.refresh(role)
|
|
return role
|
|
|
|
|
|
@router.put("/{role_id}", response_model=RoleOut)
|
|
def update_role(role_id: int, payload: RoleUpdate, db: Session = Depends(get_db)):
|
|
role = db.query(Role).filter(Role.role_id == role_id).first()
|
|
if not role:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Role not found")
|
|
if payload.role_name is not None:
|
|
role.role_name = payload.role_name
|
|
if payload.status is not None:
|
|
role.status = payload.status
|
|
if payload.remark is not None:
|
|
role.remark = payload.remark
|
|
db.commit()
|
|
db.refresh(role)
|
|
return role
|
|
|
|
|
|
@router.delete("/{role_id}")
|
|
def delete_role(role_id: int, db: Session = Depends(get_db)):
|
|
role = db.query(Role).filter(Role.role_id == role_id).first()
|
|
if not role:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Role not found")
|
|
|
|
# 检查是否有用户关联了该角色
|
|
user_count = db.query(User).filter(
|
|
User.role_ids.contains([role_id]),
|
|
User.is_deleted == 0
|
|
).count()
|
|
if user_count > 0:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Cannot delete role with assigned users")
|
|
|
|
# Delete permissions
|
|
db.query(RolePermission).filter(RolePermission.role_id == role_id).delete()
|
|
|
|
db.delete(role)
|
|
db.commit()
|
|
return {"status": "ok"}
|
|
|
|
|
|
@router.get("/{role_id}/permissions")
|
|
def get_role_permissions(role_id: int, db: Session = Depends(get_db)):
|
|
rows = db.query(RolePermission.perm_id).filter(RolePermission.role_id == role_id).all()
|
|
return [r[0] for r in rows]
|
|
|
|
|
|
@router.put("/{role_id}/permissions")
|
|
def update_role_permissions(role_id: int, payload: RolePermissionUpdate, db: Session = Depends(get_db)):
|
|
exists = db.query(Role).filter(Role.role_id == role_id).first()
|
|
if not exists:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Role not found")
|
|
|
|
db.query(RolePermission).filter(RolePermission.role_id == role_id).delete()
|
|
for perm_id in payload.perm_ids:
|
|
db.add(RolePermission(role_id=role_id, perm_id=perm_id))
|
|
db.commit()
|
|
return {"status": "ok"}
|
|
|
|
|
|
@router.get("/{role_id}/users")
|
|
def get_role_users(role_id: int, db: Session = Depends(get_db)):
|
|
from app.models import User
|
|
users = (
|
|
db.query(User)
|
|
.join(UserRole, UserRole.user_id == User.user_id)
|
|
.filter(UserRole.role_id == role_id, User.is_deleted == 0)
|
|
.all()
|
|
)
|
|
return users |