shanxia/bx/bx-backend/app/services/role.py

from sqlalchemy.orm import Session

from app.models.menu import Menu
from app.models.role import Role
from app.models.user import User
from app.services.base import CrudService


class RoleService:

    @staticmethod
    def get_by_id(db: Session, id: int) -> Role | None:
        return CrudService.get_by_id(Role, db, id)

    @staticmethod
    def get_paged(db: Session, page: int, page_size: int, filters: list | None = None):
        return CrudService.get_paged(Role, db, page, page_size, filters)

    @staticmethod
    def create(db: Session, obj_in: dict) -> Role:
        return CrudService.create(Role, db, obj_in)

    @staticmethod
    def update(db: Session, id: int, obj_in: dict) -> Role | None:
        return CrudService.update(Role, db, id, obj_in)

    @staticmethod
    def soft_delete(db: Session, id: int) -> bool:
        return CrudService.soft_delete(Role, db, id)

    @staticmethod
    def get_users_by_role(db: Session, role_id: int) -> list[User]:
        role = db.query(Role).filter(Role.id == role_id, Role.del_flag == 0).first()
        if not role:
            return []
        return list(role.users)

    @staticmethod
    def assign_users(db: Session, role_id: int, user_ids: list[int]) -> Role | None:
        role = db.query(Role).filter(Role.id == role_id, Role.del_flag == 0).first()
        if not role:
            return None
        users = db.query(User).filter(User.id.in_(user_ids), User.del_flag == 0).all()
        role.users = list(users)
        db.flush()
        return role

    @staticmethod
    def assign_menus(db: Session, role_id: int, menu_ids: list[int]) -> Role | None:
        """为角色分配菜单权限（全量替换）"""
        role = db.query(Role).filter(Role.id == role_id, Role.del_flag == 0).first()
        if not role:
            return None
        menus = db.query(Menu).filter(Menu.id.in_(menu_ids), Menu.del_flag == 0).all()
        role.menus = list(menus)
        db.flush()
        return role