116 lines
3.8 KiB
Python
116 lines
3.8 KiB
Python
from datetime import timedelta
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.config import settings
|
|
from app.database import get_db
|
|
from app.dependencies import (
|
|
generate_reset_token,
|
|
get_current_user,
|
|
reset_token_expiry,
|
|
)
|
|
from app.models.user import User
|
|
from app.schemas.user import (
|
|
ForgotPasswordRequest,
|
|
LoginRequest,
|
|
MessageResponse,
|
|
ResetPasswordRequest,
|
|
TokenResponse,
|
|
UserCreate,
|
|
UserResponse,
|
|
UserUpdate,
|
|
)
|
|
from app.services.user_service import (
|
|
authenticate_user,
|
|
create_password_reset_token,
|
|
create_user,
|
|
get_user_by_email,
|
|
get_user_by_username,
|
|
get_valid_reset_token,
|
|
update_user,
|
|
)
|
|
from app.utils.security import create_access_token, create_refresh_token, hash_password
|
|
|
|
router = APIRouter(prefix="/auth", tags=["auth"])
|
|
|
|
|
|
@router.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
|
|
def register(data: UserCreate, db: Session = Depends(get_db)) -> User:
|
|
if get_user_by_email(db, data.email.lower()):
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Email already registered")
|
|
|
|
if get_user_by_username(db, data.username.lower()):
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Username already taken")
|
|
|
|
return create_user(db, data)
|
|
|
|
|
|
@router.post("/login", response_model=TokenResponse)
|
|
def login(data: LoginRequest, db: Session = Depends(get_db)) -> TokenResponse:
|
|
user = authenticate_user(db, data.username_or_email, data.password)
|
|
if user is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid username/email or password",
|
|
)
|
|
|
|
if not user.is_active:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Account is inactive")
|
|
|
|
access_token = create_access_token(
|
|
user.email,
|
|
timedelta(days=7) if data.remember_me else None,
|
|
)
|
|
|
|
return TokenResponse(
|
|
access_token=access_token,
|
|
refresh_token=create_refresh_token(user.email),
|
|
user=UserResponse.model_validate(user),
|
|
)
|
|
|
|
|
|
@router.get("/me", response_model=UserResponse)
|
|
def get_me(current_user: User = Depends(get_current_user)) -> User:
|
|
return current_user
|
|
|
|
|
|
@router.patch("/me", response_model=UserResponse)
|
|
def update_me(
|
|
data: UserUpdate,
|
|
current_user: User = Depends(get_current_user),
|
|
db: Session = Depends(get_db),
|
|
) -> User:
|
|
return update_user(db, current_user, data)
|
|
|
|
|
|
@router.post("/forgot-password", response_model=MessageResponse)
|
|
def forgot_password(data: ForgotPasswordRequest, db: Session = Depends(get_db)) -> MessageResponse:
|
|
user = get_user_by_email(db, data.email.lower())
|
|
if user is None:
|
|
return MessageResponse(message="If that email exists, a reset link has been sent.")
|
|
|
|
token = generate_reset_token()
|
|
create_password_reset_token(db, user, token, reset_token_expiry())
|
|
|
|
response = MessageResponse(message="If that email exists, a reset link has been sent.")
|
|
if settings.debug:
|
|
response.reset_token = token
|
|
return response
|
|
|
|
|
|
@router.post("/reset-password", response_model=MessageResponse)
|
|
def reset_password(data: ResetPasswordRequest, db: Session = Depends(get_db)) -> MessageResponse:
|
|
reset = get_valid_reset_token(db, data.token)
|
|
if reset is None:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid or expired reset token")
|
|
|
|
user = db.query(User).filter(User.id == reset.user_id).first()
|
|
if user is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
|
|
|
|
user.hashed_password = hash_password(data.new_password)
|
|
reset.used = True
|
|
db.commit()
|
|
|
|
return MessageResponse(message="Password reset successfully. You can now log in.")
|