enviodeckauthenticationapi-s2f19c/app/api/routes/auth.py

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from slowapi import Limiter
from slowapi.util import get_remote_address
from fastapi import Request

from app.db.session import get_db
from app.schemas.auth import (
    OTPRequest,
    OTPVerify,
    EmailSignup,
    EmailLogin,
    GoogleLogin,
    AppleLogin,
    Token,
)
from app.schemas.user import UserCreate
from app.models.user import AuthProvider
from app.services.user import (
    get_user_by_phone,
    get_user_by_email,
    create_user,
    authenticate_user,
    verify_user_phone,
)
from app.utils.otp import generate_otp, store_otp, verify_otp, send_otp
from app.utils.auth import create_access_token

router = APIRouter()
limiter = Limiter(key_func=get_remote_address)


@router.post("/request-otp", response_model=dict)
@limiter.limit("5/minute")
async def request_otp(
    request: Request, otp_request: OTPRequest, db: Session = Depends(get_db)
):
    """Request OTP for phone number authentication"""

    # Generate and store OTP
    otp = generate_otp()
    if not store_otp(otp_request.phone_number, otp):
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Failed to generate OTP",
        )

    # Send OTP (mock implementation)
    if not send_otp(otp_request.phone_number, otp):
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Failed to send OTP",
        )

    return {"message": "OTP sent successfully"}


@router.post("/verify-otp", response_model=Token)
async def verify_otp_endpoint(otp_verify: OTPVerify, db: Session = Depends(get_db)):
    """Verify OTP and return access token"""

    # Verify OTP
    if not verify_otp(otp_verify.phone_number, otp_verify.otp):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid or expired OTP"
        )

    # Get or create user
    user = get_user_by_phone(db, otp_verify.phone_number)
    if not user:
        user_create = UserCreate(
            phone_number=otp_verify.phone_number, auth_provider=AuthProvider.PHONE
        )
        user = create_user(db, user_create)
    else:
        user = verify_user_phone(db, otp_verify.phone_number)

    # Create access token
    access_token = create_access_token(data={"sub": str(user.id)})

    return {"access_token": access_token, "token_type": "bearer"}


@router.post("/signup-email", response_model=Token)
async def signup_email(signup_data: EmailSignup, db: Session = Depends(get_db)):
    """Sign up with email and password"""

    # Check if user already exists
    if get_user_by_email(db, signup_data.email):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST, detail="Email already registered"
        )

    # Create user
    user_create = UserCreate(
        email=signup_data.email,
        name=signup_data.name,
        password=signup_data.password,
        auth_provider=AuthProvider.EMAIL,
    )
    user = create_user(db, user_create)

    # Create access token
    access_token = create_access_token(data={"sub": str(user.id)})

    return {"access_token": access_token, "token_type": "bearer"}


@router.post("/login-email", response_model=Token)
async def login_email(login_data: EmailLogin, db: Session = Depends(get_db)):
    """Login with email and password"""

    user = authenticate_user(db, login_data.email, login_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password",
        )

    if not user.is_active:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST, detail="Inactive user"
        )

    # Create access token
    access_token = create_access_token(data={"sub": str(user.id)})

    return {"access_token": access_token, "token_type": "bearer"}


@router.post("/login-google", response_model=Token)
async def login_google(google_data: GoogleLogin, db: Session = Depends(get_db)):
    """Login with Google OAuth token"""

    # Mock Google token verification
    # In production, verify the token with Google's API
    mock_user_data = {"email": "user@gmail.com", "name": "Google User"}

    # Get or create user
    user = get_user_by_email(db, mock_user_data["email"])
    if not user:
        user_create = UserCreate(
            email=mock_user_data["email"],
            name=mock_user_data["name"],
            auth_provider=AuthProvider.GOOGLE,
        )
        user = create_user(db, user_create)

    # Create access token
    access_token = create_access_token(data={"sub": str(user.id)})

    return {"access_token": access_token, "token_type": "bearer"}


@router.post("/login-apple", response_model=Token)
async def login_apple(apple_data: AppleLogin, db: Session = Depends(get_db)):
    """Login with Apple OAuth token"""

    # Mock Apple token verification
    # In production, verify the token with Apple's API
    mock_user_data = {"email": "user@icloud.com", "name": "Apple User"}

    # Get or create user
    user = get_user_by_email(db, mock_user_data["email"])
    if not user:
        user_create = UserCreate(
            email=mock_user_data["email"],
            name=mock_user_data["name"],
            auth_provider=AuthProvider.APPLE,
        )
        user = create_user(db, user_create)

    # Create access token
    access_token = create_access_token(data={"sub": str(user.id)})

    return {"access_token": access_token, "token_type": "bearer"}