You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
78 lines
2.8 KiB
Python
78 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session, selectinload
|
|
|
|
from iti.auth import require_permission
|
|
from iti.db import get_db
|
|
from iti.responses import ok, page
|
|
|
|
from iti_system.models import Role
|
|
from iti_system.schemas import RoleCreate, RoleQuery, RoleUpdate
|
|
from iti_system.services import apply_keyword, bind_menus, dump_role, get_or_404, paginate_query
|
|
|
|
|
|
router = APIRouter(prefix="/sys/role", tags=["system.role"])
|
|
|
|
|
|
@router.get("/list", dependencies=[Depends(require_permission("system:role:list"))])
|
|
def list_role(query: RoleQuery = Depends(), db: Session = Depends(get_db)):
|
|
return ok([dump_role(role) for role in db.scalars(_query_roles(query)).all()])
|
|
|
|
|
|
@router.get("/page", dependencies=[Depends(require_permission("system:role:list"))])
|
|
def page_role(query: RoleQuery = Depends(), db: Session = Depends(get_db)):
|
|
result = paginate_query(db, _query_roles(query), page=query.page, size=query.size)
|
|
return page([dump_role(role) for role in result["items"]], result["page"])
|
|
|
|
|
|
@router.post("", dependencies=[Depends(require_permission("system:role:create"))])
|
|
def create_role(payload: RoleCreate, db: Session = Depends(get_db)):
|
|
role = Role(
|
|
name=payload.name,
|
|
code=payload.code,
|
|
desc=payload.desc,
|
|
sort=payload.sort,
|
|
status=payload.status,
|
|
menus=bind_menus(db, payload.permissions),
|
|
)
|
|
db.add(role)
|
|
db.commit()
|
|
db.refresh(role)
|
|
return ok(dump_role(role))
|
|
|
|
|
|
@router.put("/{id}", dependencies=[Depends(require_permission("system:role:edit"))])
|
|
def update_role(id: str, payload: RoleUpdate, db: Session = Depends(get_db)):
|
|
role = get_or_404(db, Role, id, "角色不存在")
|
|
values = payload.model_dump(exclude_unset=True)
|
|
for key in ("name", "code", "desc", "sort", "status"):
|
|
if key in values:
|
|
setattr(role, key, values[key])
|
|
if values.get("permissions") is not None:
|
|
role.menus = bind_menus(db, values["permissions"])
|
|
db.commit()
|
|
db.refresh(role)
|
|
return ok(dump_role(role))
|
|
|
|
|
|
@router.delete("/{id}", dependencies=[Depends(require_permission("system:role:delete"))])
|
|
def delete_role(id: str, db: Session = Depends(get_db)):
|
|
role = get_or_404(db, Role, id, "角色不存在")
|
|
db.delete(role)
|
|
db.commit()
|
|
return ok()
|
|
|
|
|
|
def _query_roles(query: RoleQuery):
|
|
stmt = select(Role).options(selectinload(Role.menus)).order_by(Role.sort, Role.created_at.desc())
|
|
stmt = apply_keyword(stmt, query.keyword, Role.name, Role.code)
|
|
if query.name:
|
|
stmt = stmt.where(Role.name.like(f"%{query.name}%"))
|
|
if query.code:
|
|
stmt = stmt.where(Role.code.like(f"%{query.code}%"))
|
|
if query.status:
|
|
stmt = stmt.where(Role.status == query.status)
|
|
return stmt
|