✨ Update code for login API (#571)

* ✨ Update models for login API

* ✨ Add authenticate simplified CRUD

* ♻️ Refactor get_current_user dependency, integrate is_active

* ✨ 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 import crud
-from app.api import deps
+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 @@ from app.utils import (
 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()
+    session: SessionDep, form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
-) -> Any:
+) -> Token:
     """
     OAuth2 compatible token login, get an access token for future requests
     """
-    user = crud.user.authenticate(
+    user = crud.authenticate(
-        db, email=form_data.username, password=form_data.password
+        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 {
+    return Token(
-        "access_token": security.create_access_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)
+@router.post("/login/test-token", response_model=UserOut)
-def test_token(current_user: models.User = Depends(deps.get_current_user)) -> Any:
+def test_token(current_user: CurrentUser) -> Any:
     """
     Test access token
     """
     return current_user
 
 
-@router.post("/password-recovery/{email}", response_model=schemas.Msg)
+@router.post("/password-recovery/{email}")
-def recover_password(email: str, db: Session = Depends(deps.get_db)) -> Any:
+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(...),
+    session: SessionDep,
-    new_password: str = Body(...),
+    body: NewPassword,
-    db: Session = Depends(deps.get_db),
+) -> Message:
-) -> Any:
     """
     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)
+    session.add(user)
-    db.commit()
+    session.commit()
-    return {"msg": "Password updated successfully"}
+    return Message(message="Password updated successfully")

src/backend/app/app/api/deps.py

@@ -39,18 +39,12 @@ def get_current_user(session: SessionDep, token: TokenDep) -> User:
     user = session.get(User, token_data.sub)
     if not user:
         raise HTTPException(status_code=404, detail="User not found")
+    if not user.is_active:
+        raise HTTPException(status_code=400, detail="Inactive user")
     return user
 
 
-def get_current_active_user(
+CurrentUser = Annotated[User, Depends(get_current_user)]
-    current_user: Annotated[User, Depends(get_current_user)]
-) -> User:
-    if not current_user.is_active:
-        raise HTTPException(status_code=400, detail="Inactive user")
-    return current_user
-
-
-CurrentUser = Annotated[User, Depends(get_current_active_user)]
 
 
 def get_current_active_superuser(current_user: CurrentUser) -> User:

src/backend/app/app/crud/__init__.py

@@ -9,7 +9,7 @@ from .crud_user import user
 
 # item = CRUDBase[Item, ItemCreate, ItemUpdate](Item)
 from sqlmodel import Session, select
-from app.core.security import get_password_hash
+from app.core.security import get_password_hash, verify_password
 from app.models import UserCreate, User
 
 
@@ -27,3 +27,12 @@ def get_user_by_email(*, session: Session, email: str) -> User | None:
     statement = select(User).where(User.email == email)
     session_user = session.exec(statement).first()
     return session_user
+
+
+def authenticate(*, session: Session, email: str, password: str) -> User | None:
+    user = get_user_by_email(session=session, email=email)
+    if not user:
+        return None
+    if not verify_password(password, user.hashed_password):
+        return None
+    return user

src/backend/app/app/models.py

@@ -73,16 +73,20 @@ class ItemOut(ItemBase):
 
 
 # Generic message
-class Msg(BaseModel):
+class Message(BaseModel):
-    msg: str
+    message: str
 
 
 # JSON payload containing access token
 class Token(BaseModel):
     access_token: str
-    token_type: str
+    token_type: str = "bearer"
 
 
 # Contents of JWT token
 class TokenPayload(BaseModel):
     sub: Union[int, None] = None
+
+class NewPassword(BaseModel):
+    token: str
+    new_password: str