42 lines
1.3 KiB
Python
42 lines
1.3 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from fastapi.responses import JSONResponse
|
|
from fastapi.security import OAuth2PasswordRequestForm
|
|
|
|
from bsition.api.models.user import User
|
|
from bsition.api.utils.jwt import write_token
|
|
from bsition.api.utils.password import verify_password
|
|
from bsition.api.utils.security import get_current_user, oauth2_scheme
|
|
from bsition.backend.postgres.users import get_user_by_username
|
|
from bsition.backend.redis.tokens import add_token, clean_tokens, remove_token
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("/token")
|
|
def login(form: OAuth2PasswordRequestForm = Depends()):
|
|
user = get_user_by_username(form.username)
|
|
if user is None or not verify_password(form.password, user[2]):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="User not found.",
|
|
)
|
|
|
|
token, expire = write_token({"sub": form.username + ":" + str(user[0])})
|
|
add_token(token, form.username, expire)
|
|
clean_tokens(form.username)
|
|
return JSONResponse(
|
|
content={
|
|
"access_token": token,
|
|
"token_type": "bearer",
|
|
},
|
|
status_code=202,
|
|
)
|
|
|
|
|
|
@router.delete("/token")
|
|
def logout(
|
|
token: str = Depends(oauth2_scheme), user: tuple = Depends(get_current_user)
|
|
):
|
|
remove_token(user[1], token)
|
|
return JSONResponse(content={"detail": "Token deleted."}, status_code=202)
|