fastapi+vue实现按钮级别的权限控制

一、前端部分

1.1 自定义指令

import store from '@/store'

// 判断是否有权限
const hasPermission = (value, el) => {
  // 检查是否配置了权限参数
  if (!Array.isArray(value) || value.length === 0) {
    throw new Error(`v-permission 需要配置权限，例如 v-permission="['xxx']"`)
  }

  // 获取用户权限，登录后从store中获取到
  const ruleNames = store.getters['permissions'] || []
  if (!Array.isArray(ruleNames)) {
    console.warn('权限数据 "menu/getRuleNames" 格式不正确，请检查 store 配置。')
    return
  }

  // 判断是否有权限
  const hasAuth = value.some((val) => ruleNames.includes(val))
  if (!hasAuth) {
    el.style.display = 'none'
  }
  return hasAuth
}

export default {
  install(Vue) {
    Vue.directive('permission', {
      bind(el, binding) {
        hasPermission(binding.value, el)
      },
      updated(el, binding) {
        hasPermission(binding.value, el)
      }
    })
  }
}

1.2 注册自定义指令

import permission from '@/utils/utils'


// 注册自定义指令
Vue.use(permission)

1.3 在组件中还用自定义指令

    <el-button v-permission="['add_dept']" type="primary" size="medium" @click="addDeptBtn">新增部门</el-button>

二、后端部分

整理思路为：用户登录后生成token，然后根据fastapi的oauth2编写依赖项，并将其注入到所有的路由函数中表示需要token才能进行访问，然后再在每一个接口函数中，使用依赖性注入判断权限标识的方法，判断根据token中的用户id是否存在此接口的权限标识

2.1 生成token的方法

def create_token(payload: dict, expires: timedelta = None):
    """
    根据用户的电话号码和密码生成token
    :param payload: 载荷-用户的电话号码和密码
    :param expires: 过期时间
    :return: token
    """
    if expires:
        expire = datetime.now() + expires
    else:
        expire = datetime.now() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    payload.update({"exp": expire})
    token = jwt.encode(payload, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    return token

2.2 登录接口需要的一些工具方法

from datetime import datetime

from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession
from starlette import status
from starlette.exceptions import HTTPException
from starlette.requests import Request
from apps.vadmin.auth.models import VadminUser, VadminRole
from apps.vadmin.auth.schemas.auth import LoginSchema
from apps.vadmin.auth.schemas.role import RoleOutSchema
from apps.vadmin.record.models import VadminLoginRecord, VadminRecordAction
from apps.vadmin.record.schemas.login import LoginForm


# 获取用户信息的函数
async def get_user_by_telephone(username: str, db: AsyncSession):
    stmt = select(VadminUser).where(VadminUser.telephone == username).filter(VadminUser.is_delete == False)
    return await db.scalar(stmt)


# 获取用户权限
async def get_user_permissions(user: VadminUser, db: AsyncSession):
    permissions = []
    user_role_list = user.roles
    for role in user_role_list:
        role_stmt = select(VadminRole).where(VadminRole.id == role.id, VadminRole.is_delete == False)
        role_model = await db.scalar(role_stmt)
        if not role_model:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail='当前用户的角色不存在！')
        role_info = RoleOutSchema.from_orm(role_model).model_dump()
        permissions.extend(menu.get('perms') for menu in role_info.get('menus', []))
    return permissions


# 记录登录操作
async def create_login_record(data: LoginSchema, request: Request, response_data: dict, db: AsyncSession):
    login_form = LoginForm(telephone=data.username, password=data.password, method="0", platform="0")
    await VadminLoginRecord.create_login_record(db, login_form, True, request, response_data)

    await VadminRecordAction.create_action_record(
        db=db,
        action_type='登录操作',
        action_user=data.username,
        action_tag='登录模块',
        action_description='用户登录',
        data={'telephone': data.username},
        req=request,
        resp=response_data,
        method=request.method,
        status=True
    )

    # 更新登录时间
    login_stmt = update(VadminUser).where(VadminUser.telephone == data.username).values(last_login=datetime.now())
    await db.execute(login_stmt)

2.2 校验token并获取token中的用户信息

from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from starlette import status
from starlette.requests import Request
from apps.vadmin.auth.models import VadminUser
from apps.vadmin.auth.schemas.auth import LoginSchema
from apps.vadmin.auth.schemas.user import OutUsrSchema
from apps.vadmin.auth.validate.auth import get_user_by_telephone, get_user_permissions, create_login_record
from core.database import db_getter
from core.utils import create_token
from utils.response import ErrorResponse

app = APIRouter()


@app.post('/login', summary="登录")
async def login(request: Request, data: LoginSchema, db: AsyncSession = Depends(db_getter)):
    # 1. 校验用户是否存在
    user = await get_user_by_telephone(data.username, db)
    if not user or not VadminUser.verify_password(data.password, user.password):
        return ErrorResponse(status_code=status.HTTP_401_UNAUTHORIZED, msg="手机号或密码错误！！！")
    # 2. 校验用户状态
    if not data.is_active:
        return ErrorResponse(status_code=status.HTTP_403_FORBIDDEN, msg="该用户已被禁用，请联系管理员！！！")
    if not data.is_staff:
        return ErrorResponse(status_code=status.HTTP_403_FORBIDDEN, msg="该用户无权限，请联系管理员！！！")

    # 3. 整理用户信息并返回
    permissions = await get_user_permissions(user, db)

    # 4. 生成token
    token = create_token({'telephone': data.username, 'user_id': user.id})

    # 5. 返回用户信息和 token
    user_info = OutUsrSchema.from_orm(user).model_dump()
    user_info.update({'permissions': permissions})
    response_data = {'user': user_info, 'token': token}
    response = {"code": 200, "data": response_data, "message": "登录成功！！！"}

    await create_login_record(data, request, response_data, db)
    return response

2.3 装饰器校验权限

# 权限依赖项
def check_permissions(required_roles: List):
    def permission_dependency(user: Dict = Depends(get_current_user)):
        user_permissions_list = user.get('permissions', [])
        for required_role in required_roles:
            if required_role not in user_permissions_list:
                raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail='您无权限操作此权限！')
        return user

    return permission_dependency

2.4 使用依赖注入校验接口是否有权限

@app.get('/dept', summary='获取部门列表', response_model=DeptSimpleResponse,
         dependencies=[Depends(check_permissions(['get_dept']))])
async def get_dept_list(params: DeptQuerySchema = Depends(), db: AsyncSession = Depends(db_getter)):
    dept_data_list = await get_dept_tree_or_list_curd(params, db)
    return SuccessResponse(data=[dept_data.model_dump() for dept_data in dept_data_list], msg='获取部门列表成功！')


@app.post('/dept', summary='创建部门', response_model=DeptSimpleResponse,
          dependencies=[Depends(check_permissions(['add_dept']))])
async def create_dept(data: DeptCreateSchema, db: AsyncSession = Depends(db_getter)):
    new_dept = await create_dept_curd(data, db)
    return SuccessResponse(data=new_dept.model_dump(), msg='创建部门成功！')


@app.put('/dept/{dept_id}', summary='更新部门信息', response_model=DeptSimpleResponse,
         dependencies=[Depends(check_permissions(['update_dept']))])
async def update_dept(dept_id: int, data: DeptUpdateSchema, db: AsyncSession = Depends(db_getter)):
    updated_dept = await update_dept_curd(dept_id, data, db)
    return SuccessResponse(data=updated_dept.model_dump(), msg='更新部门成功！')


@app.delete('/dept/{dept_id}', summary='删除部门', response_model=DeptSimpleResponse,
            dependencies=[Depends(check_permissions(['delete_dept']))])
async def delete_dept(dept_id: int, db: AsyncSession = Depends(db_getter)):
    await delete_dept_curd(dept_id, db)
    return SuccessResponse(msg='删除部门成功！')