一、前端部分
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() + expireselse: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 token2.2 登录接口需要的一些工具方法
from datetime import datetimefrom 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.rolesfor 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 ErrorResponseapp = 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. 生成tokentoken = create_token({'telephone': data.username, 'user_id': user.id})# 5. 返回用户信息和 tokenuser_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 response2.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 userreturn permission_dependency2.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='删除部门成功！')
