Coverage for api/account/controller.py: 92.59%

1from core.redis import get_redis 

2from core.dependencies import get_db 

3from core.security import verify_token 

4from sqlalchemy.ext.asyncio import AsyncSession 

5from fastapi import APIRouter, Depends, HTTPException, Response 

6from .schema import UserProfile, UserUpdate, PasswordChange 

7from utils.response import APIResponse, parse_responses, common_responses 

8from .services import get_user_by_id, update_user_profile, change_password 

9from utils.custom_exception import AuthenticationException, NotFoundException 

10from extensions.smtp import get_mailer, SMTPMailer 

11 

12router = APIRouter(tags=["Account"]) 

13 

14@router.get( 

15 "/profile", 

16 response_model=APIResponse[UserProfile], 

17 summary="Get current user profile", 

18 responses=parse_responses({ 

19 200: ("User profile retrieved successfully", UserProfile) 

20 }, common_responses) 

21) 

22async def get_user_profile_api( 

23 token: dict = Depends(verify_token), 

24 db: AsyncSession = Depends(get_db) 

25): 

26 """ 

27 Get the current authenticated user's profile information. 

28 """ 

29 try: 

30 user_id = token.get("sub") 

31 user = await get_user_by_id(db, user_id) 

32 

33 if not user: 

34 raise NotFoundException("User not found") 

35 

36 user_data = UserProfile( 

37 id=user.id, 

38 first_name=user.first_name, 

39 last_name=user.last_name, 

40 email=user.email, 

41 phone=user.phone, 

42 status=user.status, 

43 created_at=user.created_at 

44 ) 

45 

46 return APIResponse(code=200, message="User profile retrieved successfully", data=user_data) 

47 except NotFoundException: 

48 raise HTTPException(status_code=404, detail="User not found") 

49 except Exception: 

50 raise HTTPException(status_code=500) 

51 

52@router.put( 

53 "/profile", 

54 response_model=APIResponse[UserProfile], 

55 response_model_exclude_unset=True, 

56 summary="Update current user profile", 

57 responses=parse_responses({ 

58 200: ("User profile updated successfully", UserProfile), 

59 202: ("Email verification required", UserProfile) 

60 }, common_responses) 

61) 

62async def update_user_profile_api( 

63 user_update: UserUpdate, 

64 response: Response, 

65 token: dict = Depends(verify_token), 

66 db: AsyncSession = Depends(get_db), 

67 redis_client = Depends(get_redis), 

68 mailer: SMTPMailer = Depends(get_mailer) 

69): 

70 """ 

71 Update the current authenticated user's profile information (excluding password). 

72 """ 

73 try: 

74 user_id = token.get("sub") 

75 result = await update_user_profile( 

76 db, user_id, user_update, mailer, redis_client 

77 ) 

78 

79 if not result: 

80 raise NotFoundException("User not found") 

81 

82 user, email_change_requested = result 

83 

84 user_data = UserProfile( 

85 id=user.id, 

86 first_name=user.first_name, 

87 last_name=user.last_name, 

88 email=user.email, 

89 phone=user.phone, 

90 status=user.status, 

91 created_at=user.created_at 

92 ) 

93 

94 if email_change_requested: 

95 response.status_code = 202 

96 return APIResponse(code=202, message="Email verification required", data=user_data) 

97 

98 return APIResponse(code=200, message="User profile updated successfully", data=user_data) 

99 except NotFoundException: 

100 raise HTTPException(status_code=404, detail="User not found") 

101 except ValueError: 

102 raise HTTPException(status_code=409, detail="Email already exists") 

103 except Exception: 

104 raise HTTPException(status_code=500) 

105 

106@router.put( 

107 "/password", 

108 response_model=APIResponse[None], 

109 response_model_exclude_unset=True, 

110 summary="Change current user password", 

111 responses=parse_responses({ 

112 200: ("Password changed successfully", None), 

113 401: ("Invalid or expired token / Current password is incorrect", None) 

114 }, common_responses) 

115) 

116async def change_user_password_api( 

117 password_change: PasswordChange, 

118 token: dict = Depends(verify_token), 

119 db: AsyncSession = Depends(get_db), 

120 redis_client = Depends(get_redis) 

121): 

122 """ 

123 Change the current authenticated user's password. 

124 """ 

125 try: 

126 user_id = token.get("sub") 

127 success = await change_password(db, user_id, password_change, redis_client) 

128 

129 if success: 

130 return APIResponse(code=200, message="Password changed successfully") 

131 

132 except AuthenticationException: 

133 raise HTTPException(status_code=401, detail="Current password is incorrect") 

134 except Exception as e: 

135 raise HTTPException(status_code=500, detail=str(e))