from sqlalchemy.orm import Session from app.models.menu import Menu from app.schemas.menu import MenuOut from app.services.base import CrudService class MenuService: @staticmethod def get_by_id(db: Session, id: int) -> Menu | None: return CrudService.get_by_id(Menu, db, id) @staticmethod def get_paged(db: Session, page: int, page_size: int, filters: list | None = None): return CrudService.get_paged(Menu, db, page, page_size, filters) @staticmethod def create(db: Session, obj_in: dict) -> Menu: return CrudService.create(Menu, db, obj_in) @staticmethod def update(db: Session, id: int, obj_in: dict) -> Menu | None: return CrudService.update(Menu, db, id, obj_in) @staticmethod def soft_delete(db: Session, id: int) -> bool: return CrudService.soft_delete(Menu, db, id) @staticmethod def get_menu_tree(db: Session) -> list[MenuOut]: """管理端菜单树。relationship.children 未带 del_flag,joinedload 会把已软删子节点一并查出; 这里只查 del_flag=0 的记录并组装为 MenuOut,避免界面删除成功仍显示旧数据,且不修改 ORM 以免 commit 误写。""" all_alive = ( db.query(Menu) .filter(Menu.del_flag == 0) .order_by(Menu.sort_order) .all() ) by_parent: dict[int, list[Menu]] = {} for m in all_alive: by_parent.setdefault(m.parent_id, []).append(m) def _to_out(m: Menu, children: list[MenuOut] | None) -> MenuOut: return MenuOut( id=m.id, parent_id=m.parent_id, menu_name=m.menu_name, menu_code=m.menu_code, menu_type=m.menu_type, path=m.path, sort_order=m.sort_order, status=m.status, children=children, ) def _build(pid: int) -> list[MenuOut]: items = by_parent.get(pid, []) result: list[MenuOut] = [] for m in items: kids = _build(m.id) result.append(_to_out(m, kids if kids else None)) return result return _build(0) @staticmethod def get_menus_by_role(db: Session, role_id: int) -> list[Menu]: from app.models.role import Role role = db.query(Role).filter(Role.id == role_id, Role.del_flag == 0).first() if not role: return [] return list(role.menus) @staticmethod def get_user_menu_tree(db: Session, user) -> list[MenuOut]: """根据用户角色构建可访问的菜单树""" # 收集用户所有角色关联的菜单 ID menu_ids: set[int] = set() for role in user.roles: for menu in role.menus: if menu.del_flag == 0 and menu.status == 1: menu_ids.add(menu.id) if not menu_ids: return [] # 向上补齐祖先菜单,保证树结构路径完整 all_ids = set(menu_ids) changed = True while changed: changed = False rows = db.query(Menu.parent_id).filter( Menu.id.in_(all_ids), Menu.parent_id != 0, Menu.del_flag == 0, ).all() for (pid,) in rows: if pid and pid not in all_ids: all_ids.add(pid) changed = True # 查询所有相关菜单 menus = ( db.query(Menu) .filter(Menu.id.in_(all_ids), Menu.del_flag == 0, Menu.status == 1) .order_by(Menu.sort_order) .all() ) # 构建树形结构 def _to_out(m: Menu, children: list[MenuOut] | None = None) -> MenuOut: return MenuOut( id=m.id, parent_id=m.parent_id, menu_name=m.menu_name, menu_code=m.menu_code, menu_type=m.menu_type, path=m.path, sort_order=m.sort_order, status=m.status, children=children, ) # 按 parent_id 分组 children_map: dict[int, list[Menu]] = {} for m in menus: children_map.setdefault(m.parent_id, []).append(m) def _build(pid: int) -> list[MenuOut]: items = children_map.get(pid, []) return [_to_out(m, _build(m.id)) for m in items] return _build(0)