当前位置: 首页 > news >正文

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='删除部门成功!')

相关文章:

  • Golang学习笔记_33——桥接模式
  • oracle between and包含边界
  • 罗德与施瓦茨ZNB20,矢量网络分析仪9KHz-20GHz
  • Redis集群主从切换源码解读
  • MySQL 学习笔记:从基础到进阶
  • 格瑞普推出革命性半固态电池,为行业无人机续航注入未来动力
  • ib网络状态探测
  • 低代码系统-产品架构案例介绍、某PaaS(十四)
  • 如何简单的去使用jconsloe 查看线程 (多线程编程篇1)
  • 深度学习和机器学习的本质区别(白话版)
  • 基于FPGA的制冷型红外成像电路设计(论文+图纸)
  • 基于SpringBoot+vue+uniapp的投票小程序+LW示例参考
  • docker删除镜像详细教程
  • DeepSeek的本地化部署
  • 【达梦数据库】dblink连接[SqlServer/Mysql]报错处理
  • 百度搜索融合 DeepSeek 满血版,开启智能搜索新篇
  • SVM对偶问题
  • Redis 监视器:深入解析与实战指南
  • HomeAssistant 发现MQTT设备(温度,湿度,开关)
  • DelayQueue实现原理
  • 西安市未央区委书记刘国荣已任西咸新区党工委书记
  • 视频|王弘治:王太后,“先天宫斗圣体”?
  • 中国巴西民间推动建立经第三方验证的“森林友好型”牛肉供应链
  • 欧元区财长会讨论国际形势及应对美国关税政策
  • 江西吉水通报一男子拒服兵役:不得考公,两年内经商、升学等受限
  • 上海护师邢红获第50届南丁格尔奖,她为何能摘得护理界最高荣誉