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

一、前端部分

1.1 自界说指令

1. import store from '@/store'
3. // 判断是否有权限
4. const hasPermission = (value, el) => {
5.   // 检查是否配置了权限参数
6.   if (!Array.isArray(value) || value.length === 0) {
7.     throw new Error(`v-permission 需要配置权限，例如 v-permission="['xxx']"`)
8.   }
10.   // 获取用户权限，登录后从store中获取到
11.   const ruleNames = store.getters['permissions'] || []
12.   if (!Array.isArray(ruleNames)) {
13.     console.warn('权限数据 "menu/getRuleNames" 格式不正确，请检查 store 配置。')
14.     return
15.   }
17.   // 判断是否有权限
18.   const hasAuth = value.some((val) => ruleNames.includes(val))
19.   if (!hasAuth) {
20.     el.style.display = 'none'
21.   }
22.   return hasAuth
23. }
25. export default {
26.   install(Vue) {
27.     Vue.directive('permission', {
28.       bind(el, binding) {
29.         hasPermission(binding.value, el)
30.       },
31.       updated(el, binding) {
32.         hasPermission(binding.value, el)
33.       }
34.     })
35.   }
36. }

复制代码

1.2 注册自界说指令

1. import permission from '@/utils/utils'
4. // 注册自定义指令
5. Vue.use(permission)

复制代码

1.3 在组件中还用自界说指令

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

复制代码

二、后端部分

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

1. def create_token(payload: dict, expires: timedelta = None):
2.     """
3.     根据用户的电话号码和密码生成token
4.     :param payload: 载荷-用户的电话号码和密码
5.     :param expires: 过期时间
6.     :return: token
7.     """
8.     if expires:
9.         expire = datetime.now() + expires
10.     else:
11.         expire = datetime.now() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
12.     payload.update({"exp": expire})
13.     token = jwt.encode(payload, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
14.     return token

复制代码

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

1. from datetime import datetime
3. from sqlalchemy import select, update
4. from sqlalchemy.ext.asyncio import AsyncSession
5. from starlette import status
6. from starlette.exceptions import HTTPException
7. from starlette.requests import Request
8. from apps.vadmin.auth.models import VadminUser, VadminRole
9. from apps.vadmin.auth.schemas.auth import LoginSchema
10. from apps.vadmin.auth.schemas.role import RoleOutSchema
11. from apps.vadmin.record.models import VadminLoginRecord, VadminRecordAction
12. from apps.vadmin.record.schemas.login import LoginForm
15. # 获取用户信息的函数
16. async def get_user_by_telephone(username: str, db: AsyncSession):
17.     stmt = select(VadminUser).where(VadminUser.telephone == username).filter(VadminUser.is_delete == False)
18.     return await db.scalar(stmt)
21. # 获取用户权限
22. async def get_user_permissions(user: VadminUser, db: AsyncSession):
23.     permissions = []
24.     user_role_list = user.roles
25.     for role in user_role_list:
26.         role_stmt = select(VadminRole).where(VadminRole.id == role.id, VadminRole.is_delete == False)
27.         role_model = await db.scalar(role_stmt)
28.         if not role_model:
29.             raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail='当前用户的角色不存在！')
30.         role_info = RoleOutSchema.from_orm(role_model).model_dump()
31.         permissions.extend(menu.get('perms') for menu in role_info.get('menus', []))
32.     return permissions
35. # 记录登录操作
36. async def create_login_record(data: LoginSchema, request: Request, response_data: dict, db: AsyncSession):
37.     login_form = LoginForm(telephone=data.username, password=data.password, method="0", platform="0")
38.     await VadminLoginRecord.create_login_record(db, login_form, True, request, response_data)
40.     await VadminRecordAction.create_action_record(
41.         db=db,
42.         action_type='登录操作',
43.         action_user=data.username,
44.         action_tag='登录模块',
45.         action_description='用户登录',
46.         data={'telephone': data.username},
47.         req=request,
48.         resp=response_data,
49.         method=request.method,
50.         status=True
51.     )
53.     # 更新登录时间
54.     login_stmt = update(VadminUser).where(VadminUser.telephone == data.username).values(last_login=datetime.now())
55.     await db.execute(login_stmt)

复制代码

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

1. from fastapi import APIRouter, Depends
2. from sqlalchemy.ext.asyncio import AsyncSession
3. from starlette import status
4. from starlette.requests import Request
5. from apps.vadmin.auth.models import VadminUser
6. from apps.vadmin.auth.schemas.auth import LoginSchema
7. from apps.vadmin.auth.schemas.user import OutUsrSchema
8. from apps.vadmin.auth.validate.auth import get_user_by_telephone, get_user_permissions, create_login_record
9. from core.database import db_getter
10. from core.utils import create_token
11. from utils.response import ErrorResponse
13. app = APIRouter()
16. @app.post('/login', summary="登录")
17. async def login(request: Request, data: LoginSchema, db: AsyncSession = Depends(db_getter)):
18.     # 1. 校验用户是否存在
19.     user = await get_user_by_telephone(data.username, db)
20.     if not user or not VadminUser.verify_password(data.password, user.password):
21.         return ErrorResponse(status_code=status.HTTP_401_UNAUTHORIZED, msg="手机号或密码错误！！！")
22.     # 2. 校验用户状态
23.     if not data.is_active:
24.         return ErrorResponse(status_code=status.HTTP_403_FORBIDDEN, msg="该用户已被禁用，请联系管理员！！！")
25.     if not data.is_staff:
26.         return ErrorResponse(status_code=status.HTTP_403_FORBIDDEN, msg="该用户无权限，请联系管理员！！！")
28.     # 3. 整理用户信息并返回
29.     permissions = await get_user_permissions(user, db)
31.     # 4. 生成token
32.     token = create_token({'telephone': data.username, 'user_id': user.id})
34.     # 5. 返回用户信息和 token
35.     user_info = OutUsrSchema.from_orm(user).model_dump()
36.     user_info.update({'permissions': permissions})
37.     response_data = {'user': user_info, 'token': token}
38.     response = {"code": 200, "data": response_data, "message": "登录成功！！！"}
40.     await create_login_record(data, request, response_data, db)
41.     return response

复制代码

2.3 装饰器校验权限

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

复制代码

2.4 利用依靠注入校验接口是否有权限

1. @app.get('/dept', summary='获取部门列表', response_model=DeptSimpleResponse,
2.          dependencies=[Depends(check_permissions(['get_dept']))])
3. async def get_dept_list(params: DeptQuerySchema = Depends(), db: AsyncSession = Depends(db_getter)):
4.     dept_data_list = await get_dept_tree_or_list_curd(params, db)
5.     return SuccessResponse(data=[dept_data.model_dump() for dept_data in dept_data_list], msg='获取部门列表成功！')
8. @app.post('/dept', summary='创建部门', response_model=DeptSimpleResponse,
9.           dependencies=[Depends(check_permissions(['add_dept']))])
10. async def create_dept(data: DeptCreateSchema, db: AsyncSession = Depends(db_getter)):
11.     new_dept = await create_dept_curd(data, db)
12.     return SuccessResponse(data=new_dept.model_dump(), msg='创建部门成功！')
15. @app.put('/dept/{dept_id}', summary='更新部门信息', response_model=DeptSimpleResponse,
16.          dependencies=[Depends(check_permissions(['update_dept']))])
17. async def update_dept(dept_id: int, data: DeptUpdateSchema, db: AsyncSession = Depends(db_getter)):
18.     updated_dept = await update_dept_curd(dept_id, data, db)
19.     return SuccessResponse(data=updated_dept.model_dump(), msg='更新部门成功！')
22. @app.delete('/dept/{dept_id}', summary='删除部门', response_model=DeptSimpleResponse,
23.             dependencies=[Depends(check_permissions(['delete_dept']))])
24. async def delete_dept(dept_id: int, db: AsyncSession = Depends(db_getter)):
25.     await delete_dept_curd(dept_id, db)
26.     return SuccessResponse(msg='删除部门成功！')

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。