Piotr Domański
e2af49046c
All checks were successful
Python lint and test / linttest (push) Successful in 5m30s
58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
from fastapi import HTTPException
|
|
from fastapi.security import OAuth2PasswordRequestForm
|
|
|
|
from ..auth import (
|
|
authenticate_user,
|
|
create_access_token,
|
|
create_refresh_token,
|
|
verify_refresh_token,
|
|
)
|
|
from ..domain.user import User as DBUser
|
|
from ..model.token import RefreshTokenPayload, Token
|
|
from .base import BaseController
|
|
|
|
|
|
class CreateToken(BaseController):
|
|
async def call(self, content: OAuth2PasswordRequestForm) -> Token:
|
|
async with self.async_session.begin() as session:
|
|
user = await authenticate_user(session, content.username, content.password)
|
|
|
|
if user is None:
|
|
raise HTTPException(
|
|
status_code=401, detail="Invalid username or password"
|
|
)
|
|
|
|
refresh_token = await create_refresh_token(session, user)
|
|
access_token = create_access_token(user)
|
|
|
|
return Token(
|
|
access_token=access_token,
|
|
refresh_token=refresh_token.token,
|
|
token_type="bearer",
|
|
)
|
|
|
|
|
|
class RefreshToken(BaseController):
|
|
async def call(self, content: RefreshTokenPayload) -> Token:
|
|
async with self.async_session.begin() as session:
|
|
current_token = await verify_refresh_token(session, content.refresh_token)
|
|
|
|
if current_token is None:
|
|
raise HTTPException(status_code=401, detail="Invalid token")
|
|
|
|
user = await DBUser.get(session, current_token.user_id)
|
|
|
|
if user is None:
|
|
raise HTTPException(status_code=401, detail="Invalid token")
|
|
|
|
assert user is not None
|
|
await current_token.delete(session)
|
|
|
|
refresh_token = await create_refresh_token(session, user)
|
|
access_token = create_access_token(user)
|
|
|
|
return Token(
|
|
access_token=access_token,
|
|
refresh_token=refresh_token.token,
|
|
token_type="bearer",
|
|
)
|