fooder-api/fooder/controller/token.py

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",
            )