101 lines
3.6 KiB
Python
101 lines
3.6 KiB
Python
from fastapi import APIRouter, Body, Depends, HTTPException
|
|
from sqlalchemy import delete, func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db import get_db
|
|
from app.deps import require_admin
|
|
from app.models import Permission, Role, User
|
|
from app.models.user import role_permissions, user_roles
|
|
from app.schemas.role import RoleCreate, RoleResponse, RoleUpdate
|
|
from app.services.audit_service import write_audit
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def _serialize(role: Role, user_count: int) -> RoleResponse:
|
|
return RoleResponse(
|
|
id=role.id,
|
|
code=role.code,
|
|
name=role.name,
|
|
description=role.description,
|
|
user_count=user_count,
|
|
permission_codes=[p.code for p in role.permissions],
|
|
)
|
|
|
|
|
|
@router.get("", response_model=list[RoleResponse])
|
|
def list_roles(db: Session = Depends(get_db), _: User = Depends(require_admin)) -> list[RoleResponse]:
|
|
counts = dict(db.execute(
|
|
select(user_roles.c.role_id, func.count(user_roles.c.user_id))
|
|
.group_by(user_roles.c.role_id)
|
|
).all())
|
|
roles = db.scalars(select(Role).order_by(Role.id)).all()
|
|
return [_serialize(r, counts.get(r.id, 0)) for r in roles]
|
|
|
|
|
|
@router.post("", response_model=RoleResponse)
|
|
def create_role(
|
|
payload: RoleCreate,
|
|
db: Session = Depends(get_db),
|
|
admin: User = Depends(require_admin),
|
|
) -> RoleResponse:
|
|
if db.scalar(select(Role).where(Role.code == payload.code)):
|
|
raise HTTPException(status_code=409, detail="角色 code 已存在")
|
|
role = Role(code=payload.code, name=payload.name, description=payload.description)
|
|
db.add(role)
|
|
db.flush()
|
|
_apply_permissions(db, role, payload.permission_codes)
|
|
write_audit(db, "role.create", actor_id=admin.id, target_type="role", target_id=str(role.id))
|
|
db.commit()
|
|
db.refresh(role)
|
|
return _serialize(role, 0)
|
|
|
|
|
|
@router.patch("/{role_id}", response_model=RoleResponse)
|
|
def update_role(
|
|
role_id: int,
|
|
payload: RoleUpdate,
|
|
db: Session = Depends(get_db),
|
|
admin: User = Depends(require_admin),
|
|
) -> RoleResponse:
|
|
role = db.get(Role, role_id)
|
|
if not role:
|
|
raise HTTPException(status_code=404, detail="角色不存在")
|
|
if payload.name is not None:
|
|
role.name = payload.name
|
|
if payload.description is not None:
|
|
role.description = payload.description
|
|
if payload.permission_codes is not None:
|
|
_apply_permissions(db, role, payload.permission_codes)
|
|
write_audit(db, "role.update", actor_id=admin.id, target_type="role", target_id=str(role.id))
|
|
db.commit()
|
|
db.refresh(role)
|
|
user_count = db.scalar(
|
|
select(func.count()).select_from(user_roles).where(user_roles.c.role_id == role.id)
|
|
) or 0
|
|
return _serialize(role, user_count)
|
|
|
|
|
|
@router.delete("/{role_id}")
|
|
def delete_role(
|
|
role_id: int,
|
|
db: Session = Depends(get_db),
|
|
admin: User = Depends(require_admin),
|
|
) -> dict:
|
|
role = db.get(Role, role_id)
|
|
if not role:
|
|
raise HTTPException(status_code=404, detail="角色不存在")
|
|
# 清掉关联
|
|
db.execute(delete(user_roles).where(user_roles.c.role_id == role_id))
|
|
db.execute(delete(role_permissions).where(role_permissions.c.role_id == role_id))
|
|
write_audit(db, "role.delete", actor_id=admin.id, target_type="role", target_id=str(role_id),
|
|
detail={"code": role.code, "name": role.name})
|
|
db.delete(role)
|
|
db.commit()
|
|
return {"deleted": True, "id": role_id}
|
|
|
|
|
|
def _apply_permissions(db: Session, role: Role, codes: list[str]) -> None:
|
|
permissions = db.scalars(select(Permission).where(Permission.code.in_(codes))).all() if codes else []
|
|
role.permissions = list(permissions)
|