56 lines
2.4 KiB
Python
56 lines
2.4 KiB
Python
# _*_ coding: utf-8 _*_
|
|
# @Time :2022/5/29 09:16
|
|
# @Email :508737091@qq.com
|
|
# @Author :qiangyanwen
|
|
# @File :user_api.py
|
|
|
|
from fastapi import APIRouter, Body, Depends, Query
|
|
from typing_extensions import Annotated
|
|
from utils.jwt_token import parse_token
|
|
from config import settings
|
|
from config.factory import AutomationResponse
|
|
from entity.user_entity import UserFrom, ActiveUser
|
|
from sqlalchemy.orm import Session
|
|
from config.db_session import get_db
|
|
from repository.user_repository import login_user, active_user
|
|
from service.user_service import user_service
|
|
from utils.jwt_token import create_token
|
|
from utils.response import success_200, error_211
|
|
from fastapi.security import OAuth2PasswordRequestForm
|
|
from datetime import timedelta
|
|
|
|
router = APIRouter(prefix="/api", tags=["用户模块"])
|
|
|
|
|
|
@router.post("/register/user", summary="注册用户", name="注册用户")
|
|
async def register(user: Annotated[UserFrom, Body(...)], db: Session = Depends(get_db)):
|
|
user = await user_service(db, user)
|
|
if isinstance(user, dict):
|
|
return success_200(data=user, message="用户注册成功")
|
|
return error_211(message=user)
|
|
|
|
|
|
@router.post("/login", summary="用户登录", name="用户登录")
|
|
async def login(user: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
|
|
user = await login_user(db, user.username, user.password)
|
|
if user[0]:
|
|
expire_time = timedelta(minutes=settings.ACCESS.ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
token = create_token(AutomationResponse.model_to_dict(user[1]), expire_time)
|
|
return {"code": 200, "message": "登录成功", "access_token": token,
|
|
"user": AutomationResponse.model_to_dict(user[1], "password"), "token_type": "Bearer"}
|
|
return error_211(message=user[1])
|
|
|
|
|
|
@router.post("/user/active", summary="用户启用和禁用", name="用户启用和禁用")
|
|
async def active(user: Annotated[ActiveUser, Body(...)], db: Session = Depends(get_db), user_id: int = Depends(parse_token)):
|
|
status = 0
|
|
if user.status == 0: # 正在启用
|
|
status = 1
|
|
if user.status == 1: # 正在禁用
|
|
status = 0
|
|
await active_user(db, user_id=user.id, status=status)
|
|
if status == 0: # 正在启用
|
|
return success_200(data=dict(status=1), message="禁用成功")
|
|
if status == 1: # 正在停用
|
|
return success_200(data=dict(status=0), message="启用成功")
|