(
session: SessionDep,
trans: Trans,
form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
)
| 24 | result_id_expr="id" |
| 25 | )) |
| 26 | async def local_login( |
| 27 | session: SessionDep, |
| 28 | trans: Trans, |
| 29 | form_data: Annotated[OAuth2PasswordRequestForm, Depends()] |
| 30 | ) -> Token: |
| 31 | origin_account = await sqlbot_decrypt(form_data.username) |
| 32 | origin_pwd = await sqlbot_decrypt(form_data.password) |
| 33 | user: BaseUserDTO = authenticate(session=session, account=origin_account, password=origin_pwd) |
| 34 | if not user: |
| 35 | raise HTTPException(status_code=400, detail=trans('i18n_login.account_pwd_error')) |
| 36 | if not user.oid or user.oid == 0: |
| 37 | raise HTTPException(status_code=400, detail=trans('i18n_login.no_associated_ws', msg = trans('i18n_concat_admin'))) |
| 38 | if user.status != 1: |
| 39 | raise HTTPException(status_code=400, detail=trans('i18n_login.user_disable', msg = trans('i18n_concat_admin'))) |
| 40 | if user.origin is not None and user.origin != 0: |
| 41 | raise HTTPException(status_code=400, detail=trans('i18n_login.origin_error')) |
| 42 | access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) |
| 43 | user_dict = user.to_dict() |
| 44 | return Token(access_token=create_access_token( |
| 45 | user_dict, expires_delta=access_token_expires |
| 46 | )) |
| 47 | |
| 48 | @router.post("/logout") |
| 49 | async def logout(session: SessionDep, request: Request, dto: LogoutSchema): |
nothing calls this directly
no test coverage detected