Spaces:
Running
Running
| import logging | |
| from fastapi import APIRouter, Header, HTTPException, status, Depends | |
| from app.schemas.auth import ( | |
| SignupDto, | |
| VerifySignupOtpDto, | |
| LoginDto, | |
| SendPasswordResetOtpDto, | |
| VerifyPasswordResetOtpDto, | |
| ResetPasswordDto, | |
| ResendOtpDto, | |
| ) | |
| from app.services.auth_service import auth_service | |
| from app.services.user_service import user_service | |
| from app.services.otp_service import otp_service | |
| from app.guards.jwt_guard import get_current_user | |
| router = APIRouter(tags=["Authentication"]) | |
| logger = logging.getLogger(__name__) | |
| # ============================================================ | |
| # SIGNUP ENDPOINTS | |
| # ============================================================ | |
| async def signup(signup_dto: SignupDto): | |
| """ | |
| Step 1: Initiate Signup | |
| Create account and send OTP to email or phone | |
| """ | |
| logger.info("Signup request") | |
| return await auth_service.signup(signup_dto) | |
| async def verify_signup_otp(dto: VerifySignupOtpDto): | |
| """ | |
| Step 2: Verify Signup OTP | |
| Verify OTP and complete signup. Returns JWT token. | |
| """ | |
| logger.info(f"Verify signup OTP: {dto.identifier}") | |
| return await auth_service.verify_signup_otp(dto.identifier, dto.code) | |
| # ============================================================ | |
| # LOGIN ENDPOINT | |
| # ============================================================ | |
| async def login(login_dto: LoginDto): | |
| """ | |
| User Login | |
| Authenticate with email or phone and password. Returns JWT token. | |
| """ | |
| logger.info(f"Login request: {login_dto.identifier}") | |
| return await auth_service.login(login_dto) | |
| # ============================================================ | |
| # PASSWORD RESET ENDPOINTS | |
| # ============================================================ | |
| async def send_password_reset_otp(dto: SendPasswordResetOtpDto): | |
| """ | |
| Step 1: Request Password Reset OTP | |
| Send OTP to email or phone for password reset. | |
| Does not reveal if account exists (security). | |
| """ | |
| logger.info(f"Send password reset OTP: {dto.identifier}") | |
| return await auth_service.send_password_reset_otp(dto.identifier) | |
| async def verify_password_reset_otp(dto: VerifyPasswordResetOtpDto): | |
| """ | |
| Step 2: Verify Password Reset OTP | |
| Verify OTP and get temporary reset token (10 minute validity). | |
| """ | |
| logger.info(f"Verify password reset OTP: {dto.identifier}") | |
| return await auth_service.verify_password_reset_otp(dto.identifier, dto.code) | |
| async def reset_password( | |
| reset_password_dto: ResetPasswordDto, | |
| x_reset_token: str = Header(None), | |
| ): | |
| """ | |
| Step 3: Reset Password | |
| Reset password using reset token from verify-password-reset-otp endpoint. | |
| Pass token in x-reset-token header. | |
| """ | |
| if not x_reset_token: | |
| raise HTTPException( | |
| status_code=status.HTTP_400_BAD_REQUEST, | |
| detail="Reset token is required in x-reset-token header" | |
| ) | |
| logger.info(f"Reset password: {reset_password_dto.identifier}") | |
| return await auth_service.reset_password(reset_password_dto, x_reset_token) | |
| # ============================================================ | |
| # RESEND OTP ENDPOINT | |
| # ============================================================ | |
| async def resend_otp(dto: ResendOtpDto): | |
| """ | |
| Resend OTP for Signup or Password Reset | |
| Specify purpose in request body ('signup' or 'password_reset'). | |
| """ | |
| logger.info(f"Resend OTP for {dto.purpose}: {dto.identifier}") | |
| is_expired = await otp_service.is_otp_expired(dto.identifier, dto.purpose) | |
| if not is_expired: | |
| raise HTTPException( | |
| status_code=status.HTTP_400_BAD_REQUEST, | |
| detail="OTP still valid. Please wait before requesting again." | |
| ) | |
| await otp_service.send_otp(dto.identifier, dto.purpose) | |
| return { | |
| "success": True, | |
| "message": "New OTP sent to your email/phone", | |
| } | |
| # ============================================================ | |
| # USER PROFILE ENDPOINTS | |
| # ============================================================ | |
| async def get_current_user_profile(current_user: dict = Depends(get_current_user)): | |
| """ | |
| Get Current User Profile | |
| Fetch current logged-in user profile. Only you can view your own profile. | |
| Requires: Bearer token in Authorization header | |
| """ | |
| logger.info(f"Get current user profile: {current_user.get('user_id')}") | |
| return await user_service.get_current_user_profile(current_user.get("user_id")) | |
| # ============================================================ | |
| # LOGOUT ENDPOINT | |
| # ============================================================ | |
| async def logout(current_user: dict = Depends(get_current_user)): | |
| """ | |
| User Logout | |
| Logout current user. Remove JWT token from frontend storage to complete logout. | |
| Requires: Bearer token in Authorization header | |
| """ | |
| logger.info(f"User logged out: {current_user.get('user_id')}") | |
| return { | |
| "success": True, | |
| "message": "Logout successful. Please remove your token from storage to complete logout.", | |
| } |