107 lines
3.8 KiB
Python
107 lines
3.8 KiB
Python
from datetime import datetime, timedelta
|
|
|
|
from fastapi import APIRouter
|
|
|
|
from app.controllers.user import UserController, user_controller
|
|
from app.core.ctx import CTX_USER_ID
|
|
from app.core.dependency import DependAuth
|
|
from app.models.admin import Api, Menu, Role, User
|
|
from app.schemas.base import BaseResponse, Fail, Success
|
|
from app.schemas.login import *
|
|
from app.schemas.users import UpdatePassword
|
|
from app.settings import settings
|
|
from app.utils.jwt import create_access_token
|
|
from app.utils.password import get_password_hash, verify_password
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("/access_token", summary="获取token")
|
|
async def login_access_token(credentials: CredentialsSchema) -> BaseResponse:
|
|
user: User = await user_controller.authenticate(credentials)
|
|
await user_controller.update_last_login(user.id)
|
|
access_token_expires = timedelta(minutes=settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
expire = datetime.utcnow() + access_token_expires
|
|
|
|
result = JWTOut(
|
|
access_token=create_access_token(
|
|
data=JWTPayload(
|
|
user_id=user.id,
|
|
username=user.username,
|
|
is_superuser=user.is_superuser,
|
|
exp=expire,
|
|
)
|
|
),
|
|
username=user.username,
|
|
)
|
|
|
|
return BaseResponse(code=200, data=result)
|
|
|
|
|
|
@router.get("/userinfo", summary="查看用户信息", dependencies=[DependAuth])
|
|
async def get_userinfo():
|
|
user_id = CTX_USER_ID.get()
|
|
user_obj = await user_controller.get(id=user_id)
|
|
to_dict = await user_obj.to_dict()
|
|
to_dict.pop("password")
|
|
to_dict["avatar"] = "https://avatars.githubusercontent.com/u/54677442?v=4"
|
|
|
|
return BaseResponse(code=200, data=to_dict)
|
|
|
|
|
|
@router.get("/usermenu", summary="查看用户菜单", dependencies=[DependAuth])
|
|
async def get_user_menu() -> BaseResponse:
|
|
user_id = CTX_USER_ID.get()
|
|
user_obj = await User.filter(id=user_id).first()
|
|
menus: list[Menu] = []
|
|
if user_obj.is_superuser:
|
|
menus = await Menu.all()
|
|
else:
|
|
role_objs: list[Role] = await user_obj.roles
|
|
for role_obj in role_objs:
|
|
menu = await role_obj.menus
|
|
menus.extend(menu)
|
|
menus = list(set(menus))
|
|
parent_menus: list[Menu] = []
|
|
for menu in menus:
|
|
if menu.parent_id == 0:
|
|
parent_menus.append(menu)
|
|
res = []
|
|
for parent_menu in parent_menus:
|
|
parent_menu_dict = await parent_menu.to_dict()
|
|
parent_menu_dict["children"] = []
|
|
for menu in menus:
|
|
if menu.parent_id == parent_menu.id:
|
|
parent_menu_dict["children"].append(await menu.to_dict())
|
|
res.append(parent_menu_dict)
|
|
return Success(data=res)
|
|
|
|
|
|
@router.get("/userapi", summary="查看用户API", dependencies=[DependAuth])
|
|
async def get_user_api() -> BaseResponse:
|
|
user_id = CTX_USER_ID.get()
|
|
user_obj = await User.filter(id=user_id).first()
|
|
if user_obj.is_superuser:
|
|
api_objs: list[Api] = await Api.all()
|
|
apis = [api.method.lower() + api.path for api in api_objs]
|
|
return Success(data=apis)
|
|
role_objs: list[Role] = await user_obj.roles
|
|
apis = []
|
|
for role_obj in role_objs:
|
|
api_objs: list[Api] = await role_obj.apis
|
|
apis.extend([api.method.lower() + api.path for api in api_objs])
|
|
apis = list(set(apis))
|
|
return Success(data=apis)
|
|
|
|
|
|
@router.post("/update_password", summary="更新用户密码", dependencies=[DependAuth])
|
|
async def update_user_password(req_in: UpdatePassword) -> BaseResponse:
|
|
user_controller = UserController()
|
|
user = await user_controller.get(req_in.id)
|
|
verified = verify_password(req_in.old_password, user.password)
|
|
if not verified:
|
|
return Fail(msg="旧密码验证错误！")
|
|
user.password = get_password_hash(req_in.new_password)
|
|
await user.save()
|
|
return Success(msg="修改成功")
|