neo-neo-todo/backend/src/neo_neo_todo/routes/token.py

import os
from datetime import timedelta
from typing import Annotated

import bcrypt
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from psycopg_pool import AsyncConnectionPool

from src.neo_neo_todo.models.member import select_member_by_email
from src.neo_neo_todo.models.token import Token, create_access_token
from src.neo_neo_todo.utils.database import get_pool

router = APIRouter(
    prefix="/token",
    tags=["token"],
)

try:
    ACCESS_TOKEN_EXPIRE_MINUTES = os.environ["ACCESS_TOKEN_EXPIRE_MINUTES"]
    TODO_SALT = os.environ["TODO_SALT"]
except KeyError:
    raise KeyError("Can't find access token expire in minutes")


def verify_password(plain_password: bytes, hashed_password: bytes) -> bool:
    """
    Verify user supplied password with the password hash in the database

    Return True if matches or False if not
    """
    return bcrypt.checkpw(plain_password, hashed_password)


def get_password_hash(password):
    return bcrypt.hashpw(password, TODO_SALT.encode("utf-8"))


INVALID_USERNAME_OR_PASSWORD_EXCEPTION: HTTPException = HTTPException(
    status_code=status.HTTP_400_BAD_REQUEST,
    detail="Incorrect email or password",
)


@router.post(
    "",
    status_code=status.HTTP_200_OK,  # default status code when login successful
)
async def login(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
    db_pool: AsyncConnectionPool = Depends(get_pool),
):
    """
    Login to the todo app

    If successful, return the JWT token
    """

    # username is actually email in this case
    member = await select_member_by_email(db_pool, form_data.username)

    if member is None:
        raise INVALID_USERNAME_OR_PASSWORD_EXCEPTION

    passwords_match = verify_password(
        form_data.password.encode("utf-8"),
        member.password_hash.encode("utf-8"),
    )

    if passwords_match:
        # to handle login logic here
        access_token_expires = timedelta(minutes=int(ACCESS_TOKEN_EXPIRE_MINUTES))

        access_token = create_access_token(
            data={"sub": member.email}, expires_delta=access_token_expires
        )

    else:
        raise INVALID_USERNAME_OR_PASSWORD_EXCEPTION

    return Token(access_token=access_token, token_type="bearer")