c65ff06918
1. 修复（kinit-api）：utils/cache.py 日志模块导入问题修复 2. 修复（kinit-api）：token解析失败会报错问题修复 3. 优化（kinit-api）：用户登录认证失败返回值优化 4. 优化（kinit-api）：获取redis方式统一改为redis_getter(request) 5. 优化（kini-api）：文件IO修改为异步操作 6. 优化（kinit-api）：关联创建人时将user_id改为create_user_id 7. 文档（kinit-api）：kinit-api/README.md 加入查询数据文档 8. 修复（kinit-admin）：用户无法导出问题修复 9. 优化（kinit-admin）：角色新增与编辑框使用默认父子联动 10. 更新（kinit-admin）：usePermissionStore 改为 useRouterStoreWithOut，因为此文件主要记录路由 11. 更新（kinit-admin）：取消用户信息的持久化存储，改为仅保存在pinia store共享中，并添加roles，permissions信息 12. 修复（kinit-admin）：菜单新增与编辑框，目录与菜单切换时会出现抖动问题修复 13. 优化（kinit-admin）：src\hooks\web\useTable.ts 优化删除数据方法 14. 优化（kinit-admin）：config/services.ts 新增返回403状态码时直接退出系统 15. 优化（kinit-admin）：将store中的本文件使用store调用的，改为this 16. 更新（kinit-admin）：路由拦截更新 17. 更新（kinit-api，kinit-admin，kinit-uni）：取消接口地址最后面的 /
63 lines
2.2 KiB
Python
63 lines
2.2 KiB
Python
#!/usr/bin/python
|
||
# -*- coding: utf-8 -*-
|
||
# @version : 1.0
|
||
# @Create Time : 2022/8/8 11:02
|
||
# @File : auth_util.py
|
||
# @IDE : PyCharm
|
||
# @desc : 简要说明
|
||
|
||
from datetime import datetime, timedelta
|
||
from fastapi import Request
|
||
from application import settings
|
||
import jwt
|
||
from apps.vadmin.auth import models
|
||
from core.database import redis_getter
|
||
from .validation import LoginValidation, LoginForm, LoginResult
|
||
from utils.aliyun_sms import AliyunSMS
|
||
|
||
|
||
class LoginManage:
|
||
"""
|
||
登录认证工具
|
||
"""
|
||
|
||
@LoginValidation
|
||
async def password_login(self, data: LoginForm, user: models.VadminUser, **kwargs) -> LoginResult:
|
||
"""
|
||
验证用户密码
|
||
"""
|
||
result = models.VadminUser.verify_password(data.password, user.password)
|
||
if result:
|
||
return LoginResult(status=True, msg="验证成功")
|
||
return LoginResult(status=False, msg="手机号或密码错误")
|
||
|
||
@LoginValidation
|
||
async def sms_login(self, data: LoginForm, request: Request, **kwargs) -> LoginResult:
|
||
"""
|
||
验证用户短信验证码
|
||
"""
|
||
rd = redis_getter(request)
|
||
sms = AliyunSMS(rd, data.telephone)
|
||
result = await sms.check_sms_code(data.password)
|
||
if result:
|
||
return LoginResult(status=True, msg="验证成功")
|
||
return LoginResult(status=False, msg="验证码错误")
|
||
|
||
@staticmethod
|
||
def create_token(payload: dict, expires: timedelta = None):
|
||
"""
|
||
创建一个生成新的访问令牌的工具函数。
|
||
|
||
pyjwt：https://github.com/jpadilla/pyjwt/blob/master/docs/usage.rst
|
||
jwt 博客：https://geek-docs.com/python/python-tutorial/j_python-jwt.html
|
||
|
||
#TODO 传入的时间为UTC时间datetime.datetime类型，但是在解码时获取到的是本机时间的时间戳
|
||
"""
|
||
if expires:
|
||
expire = datetime.utcnow() + expires
|
||
else:
|
||
expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
|
||
payload.update({"exp": expire})
|
||
encoded_jwt = jwt.encode(payload, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
|
||
return encoded_jwt
|