| 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", |
| |
| 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}") |
| |
| raise HTTPException( |
| status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, |
| detail="Internal server error" |
| ) |
| |
| if not user_profile: |
| |
| 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, |
| } |
|
|
|
|