68 lines
2.8 KiB
Python
68 lines
2.8 KiB
Python
# -*- coding: utf-8 -*-
|
||
# @version : 1.0
|
||
# @Creaet Time : 2021/10/24 16:44
|
||
# @File : views.py
|
||
# @IDE : PyCharm
|
||
# @desc : 安全认证视图
|
||
|
||
"""
|
||
JWT 表示 「JSON Web Tokens」。https://jwt.io/
|
||
|
||
它是一个将 JSON 对象编码为密集且没有空格的长字符串的标准。
|
||
|
||
通过这种方式，你可以创建一个有效期为 1 周的令牌。然后当用户第二天使用令牌重新访问时，你知道该用户仍然处于登入状态。
|
||
一周后令牌将会过期，用户将不会通过认证，必须再次登录才能获得一个新令牌。
|
||
|
||
我们需要安装 python-jose 以在 Python 中生成和校验 JWT 令牌：pip install python-jose[cryptography]
|
||
|
||
PassLib 是一个用于处理哈希密码的很棒的 Python 包。它支持许多安全哈希算法以及配合算法使用的实用程序。
|
||
推荐的算法是 「Bcrypt」：pip install passlib[bcrypt]
|
||
"""
|
||
|
||
from datetime import timedelta
|
||
from fastapi import APIRouter, Depends, Request
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from core.database import db_getter
|
||
from utils.response import SuccessResponse, ErrorResponse
|
||
from application import settings
|
||
from .login_manage import LoginManage
|
||
from .validation import LoginForm
|
||
from apps.vadmin.record.models import VadminLoginRecord
|
||
from apps.vadmin.auth.crud import MenuDal
|
||
from .current import full_admin, Auth
|
||
|
||
app = APIRouter()
|
||
|
||
|
||
@app.post("/login/", summary="登录")
|
||
async def login_for_access_token(request: Request, data: LoginForm, manage: LoginManage = Depends(),
|
||
db: AsyncSession = Depends(db_getter)):
|
||
if data.method == "0":
|
||
result = await manage.password_login(data, db, request)
|
||
elif data.method == "1":
|
||
result = await manage.sms_login(data, db, request)
|
||
else:
|
||
return ErrorResponse(msg="请使用正确的登录方式")
|
||
if not result.status:
|
||
resp = {"message": result.msg}
|
||
telephone = data.telephone
|
||
await VadminLoginRecord.\
|
||
create_login_record(db, telephone, result.status, request, resp)
|
||
return ErrorResponse(msg=result.msg)
|
||
|
||
user = result.user
|
||
access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
|
||
access_token = LoginManage.create_access_token(data={"sub": user.telephone}, expires_delta=access_token_expires)
|
||
resp = {
|
||
"access_token": access_token,
|
||
"token_type": "bearer",
|
||
"is_reset_password": user.is_reset_password
|
||
}
|
||
await VadminLoginRecord.create_login_record(db, user.telephone, result.status, request, resp)
|
||
return SuccessResponse(resp)
|
||
|
||
|
||
@app.get("/getMenuList/", summary="获取当前用户菜单树")
|
||
async def get_menu_list(auth: Auth = Depends(full_admin)):
|
||
return SuccessResponse(await MenuDal(auth.db).get_routers(auth.user))
|