55 lines
1.7 KiB
Python
55 lines
1.7 KiB
Python
"""菜单 code 权限守卫"""
|
|
from typing import Callable, Set
|
|
|
|
from fastapi import Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.auth import get_current_user
|
|
from app.core.database import get_db
|
|
from app.models.models import SysUserMenuScope
|
|
|
|
ADMIN_ROLES = {"admin", "super_admin"}
|
|
SUPER_ADMIN_ROLE = "super_admin"
|
|
|
|
|
|
def require_admin_or_super_admin(current_user: dict = Depends(get_current_user)):
|
|
"""运营端角色守卫：admin / super_admin 可访问。"""
|
|
role = current_user.get("role")
|
|
if role not in ADMIN_ROLES:
|
|
raise HTTPException(status_code=403, detail="需要运营人员权限")
|
|
return current_user
|
|
|
|
|
|
def get_user_menu_codes(db: Session, user_id: str) -> Set[str]:
|
|
"""读取用户授权的菜单 code 集合。"""
|
|
if not user_id:
|
|
return set()
|
|
rows = (
|
|
db.query(SysUserMenuScope.menu_code)
|
|
.filter(
|
|
SysUserMenuScope.user_id == user_id,
|
|
SysUserMenuScope.is_del == 0,
|
|
)
|
|
.all()
|
|
)
|
|
return {str(row[0]).strip() for row in rows if row and row[0]}
|
|
|
|
|
|
def require_menu_code(menu_code: str) -> Callable:
|
|
"""接口菜单权限守卫。super_admin 直接放行。"""
|
|
normalized = (menu_code or "").strip()
|
|
|
|
def _dep(
|
|
current_user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
role = current_user.get("role")
|
|
if role == SUPER_ADMIN_ROLE:
|
|
return current_user
|
|
codes = get_user_menu_codes(db, current_user.get("user_id"))
|
|
if normalized not in codes:
|
|
raise HTTPException(status_code=403, detail="无菜单访问权限")
|
|
return current_user
|
|
|
|
return _dep
|