mirror of
https://gh-proxy.org/https://github.com/klinsmaya/SQLBot.git
synced 2026-11-08 01:06:59 +08:00
43 lines
2.0 KiB
Python
43 lines
2.0 KiB
Python
from typing import Annotated
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from fastapi.security import OAuth2PasswordRequestForm
|
|
from apps.system.schemas.logout_schema import LogoutSchema
|
|
from apps.system.schemas.system_schema import BaseUserDTO
|
|
from common.core.deps import SessionDep, Trans
|
|
from common.utils.crypto import sqlbot_decrypt
|
|
from ..crud.user import authenticate
|
|
from common.core.security import create_access_token
|
|
from datetime import timedelta
|
|
from common.core.config import settings
|
|
from common.core.schemas import Token
|
|
from sqlbot_xpack.authentication.manage import logout as xpack_logout
|
|
router = APIRouter(tags=["login"], prefix="/login")
|
|
|
|
@router.post("/access-token")
|
|
async def local_login(
|
|
session: SessionDep,
|
|
trans: Trans,
|
|
form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
|
|
) -> Token:
|
|
origin_account = await sqlbot_decrypt(form_data.username)
|
|
origin_pwd = await sqlbot_decrypt(form_data.password)
|
|
user: BaseUserDTO = authenticate(session=session, account=origin_account, password=origin_pwd)
|
|
if not user:
|
|
raise HTTPException(status_code=400, detail=trans('i18n_login.account_pwd_error'))
|
|
if not user.oid or user.oid == 0:
|
|
raise HTTPException(status_code=400, detail=trans('i18n_login.no_associated_ws', msg = trans('i18n_concat_admin')))
|
|
if user.status != 1:
|
|
raise HTTPException(status_code=400, detail=trans('i18n_login.user_disable', msg = trans('i18n_concat_admin')))
|
|
if user.origin is not None and user.origin != 0:
|
|
raise HTTPException(status_code=400, detail=trans('i18n_login.origin_error'))
|
|
access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
user_dict = user.to_dict()
|
|
return Token(access_token=create_access_token(
|
|
user_dict, expires_delta=access_token_expires
|
|
))
|
|
|
|
@router.post("/logout")
|
|
async def logout(session: SessionDep, request: Request, dto: LogoutSchema):
|
|
if dto.origin != 0:
|
|
return await xpack_logout(session, request, dto)
|
|
return None |