✨ Refactor and upgrade login API code

src/backend/app/app/api/api_v1/endpoints/login.py

@@ -1,15 +1,15 @@
 from datetime import timedelta
-from typing import Any
+from typing import Annotated, Any
 
-from fastapi import APIRouter, Body, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException
 from fastapi.security import OAuth2PasswordRequestForm
-from sqlalchemy.orm import Session
 
-from app import crud, models, schemas
-from app.api import deps
+from app import crud
+from app.api.deps import CurrentUser, SessionDep
 from app.core import security
 from app.core.config import settings
 from app.core.security import get_password_hash
+from app.models import Message, NewPassword, Token, UserOut
 from app.utils import (
     generate_password_reset_token,
     send_reset_password_email,
@@ -19,43 +19,42 @@
 router = APIRouter()
 
 
-@router.post("/login/access-token", response_model=schemas.Token)
+@router.post("/login/access-token")
 def login_access_token(
-    db: Session = Depends(deps.get_db), form_data: OAuth2PasswordRequestForm = Depends()
-) -> Any:
+    session: SessionDep, form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
+) -> Token:
     """
     OAuth2 compatible token login, get an access token for future requests
     """
-    user = crud.user.authenticate(
-        db, email=form_data.username, password=form_data.password
+    user = crud.authenticate(
+        session=session, email=form_data.username, password=form_data.password
     )
     if not user:
         raise HTTPException(status_code=400, detail="Incorrect email or password")
-    elif not crud.user.is_active(user):
+    elif not user.is_active:
         raise HTTPException(status_code=400, detail="Inactive user")
     access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
-    return {
-        "access_token": security.create_access_token(
+    return Token(
+        access_token=security.create_access_token(
             user.id, expires_delta=access_token_expires
-        ),
-        "token_type": "bearer",
-    }
+        )
+    )
 
 
-@router.post("/login/test-token", response_model=schemas.User)
-def test_token(current_user: models.User = Depends(deps.get_current_user)) -> Any:
+@router.post("/login/test-token", response_model=UserOut)
+def test_token(current_user: CurrentUser) -> Any:
     """
     Test access token
     """
     return current_user
 
 
-@router.post("/password-recovery/{email}", response_model=schemas.Msg)
-def recover_password(email: str, db: Session = Depends(deps.get_db)) -> Any:
+@router.post("/password-recovery/{email}")
+def recover_password(email: str, session: SessionDep) -> Message:
     """
     Password Recovery
     """
-    user = crud.user.get_by_email(db, email=email)
+    user = crud.get_user_by_email(session=session, email=email)
 
     if not user:
         raise HTTPException(
@@ -66,31 +65,30 @@ def recover_password(email: str, db: Session = Depends(deps.get_db)) -> Any:
     send_reset_password_email(
         email_to=user.email, email=email, token=password_reset_token
     )
-    return {"msg": "Password recovery email sent"}
+    return Message(message="Password recovery email sent")
 
 
-@router.post("/reset-password/", response_model=schemas.Msg)
+@router.post("/reset-password/")
 def reset_password(
-    token: str = Body(...),
-    new_password: str = Body(...),
-    db: Session = Depends(deps.get_db),
-) -> Any:
+    session: SessionDep,
+    body: NewPassword,
+) -> Message:
     """
     Reset password
     """
-    email = verify_password_reset_token(token)
+    email = verify_password_reset_token(token=body.token)
     if not email:
         raise HTTPException(status_code=400, detail="Invalid token")
-    user = crud.user.get_by_email(db, email=email)
+    user = crud.get_user_by_email(session=session, email=email)
     if not user:
         raise HTTPException(
             status_code=404,
             detail="The user with this username does not exist in the system.",
         )
-    elif not crud.user.is_active(user):
+    elif not user.is_active:
         raise HTTPException(status_code=400, detail="Inactive user")
-    hashed_password = get_password_hash(new_password)
+    hashed_password = get_password_hash(password=body.new_password)
     user.hashed_password = hashed_password
-    db.add(user)
-    db.commit()
-    return {"msg": "Password updated successfully"}
+    session.add(user)
+    session.commit()
+    return Message(message="Password updated successfully")