File size: 2,103 Bytes
bef5e76 08df5ae bef5e76 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 | import pandas as pd
from datetime import datetime
from fastapi.responses import JSONResponse
from fastapi import APIRouter, HTTPException, status
from typing import Literal
from src.users.users import get_user, hash_password, verify_password
from src.middlewares.logging import get_logger
from pydantic import BaseModel
class ILogin(BaseModel):
"""Login request model."""
email: str
password: str
logger = get_logger("users service")
router = APIRouter(
prefix="/api",
tags=["Users"],
)
from typing import Optional, Literal
@router.post(
"/login",
# response_model=IUserProfile,
summary="Login by email and password",
description="💡Authenticates a user with email and password (non hashed) from frontend and returns user data if successful."
)
async def login(payload: ILogin):
"""
Authenticates a user and returns their data if credentials are valid.
"""
try:
user_profile:dict | None= await get_user(payload.email)
except Exception as E:
print(f"❌ login error while fetching user: {E}")
# Return generic 500 to client
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Internal server error"
)
if not user_profile:
# 404 or 401 – choose based on your security policy
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Email not found"
)
if user_profile.get("status") == "inactive":
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Account is inactive"
)
is_verified = verify_password(
password=payload.password,
hashed_password=user_profile.get("password")
)
if not is_verified:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Email or password invalid"
)
user_profile.pop("password", None)
return {
"status": "success",
"message": "success",
"data": user_profile,
}
|