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

1from fastapi import APIRouter, Depends, HTTPException, Response 

2from sqlalchemy.ext.asyncio import AsyncSession 

3 

4from core.dependencies import get_db 

5from core.redis import get_redis 

6from core.security import verify_token 

7from extensions.smtp import SMTPMailer, get_mailer 

8from utils.custom_exception import AuthenticationException, NotFoundException 

9from utils.response import APIResponse, common_responses, make_error_examples, parse_responses 

10 

11from .schema import PasswordChange, UserProfile, UserUpdate 

12from .services import change_password, get_user_by_id, update_user_profile 

13 

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

15 

16 

17def _to_user_profile(user) -> UserProfile: 

18 return UserProfile( 

19 id=user.id, 

20 first_name=user.first_name, 

21 last_name=user.last_name, 

22 email=user.email, 

23 pending_email=user.pending_email, 

24 phone=user.phone, 

25 status=user.status, 

26 created_at=user.created_at, 

27 ) 

28 

29 

30@router.get( 

31 "/profile", 

32 response_model=APIResponse[UserProfile], 

33 summary="Get current user profile", 

34 responses=parse_responses( 

35 {200: ("User profile retrieved successfully", UserProfile)}, common_responses 

36 ), 

37) 

38async def get_user_profile_api( 

39 token: dict = Depends(verify_token), db: AsyncSession = Depends(get_db) 

40): 

41 """ 

42 Get the current authenticated user's profile information. 

43 """ 

44 try: 

45 user_id = token.get("sub") 

46 user = await get_user_by_id(db, user_id) 

47 

48 if not user: 

49 raise NotFoundException("User not found") 

50 

51 user_data = _to_user_profile(user) 

52 

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

54 except NotFoundException: 

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

56 except Exception: 

57 raise HTTPException(status_code=500) 

58 

59 

60@router.put( 

61 "/profile", 

62 response_model=APIResponse[UserProfile], 

63 response_model_exclude_unset=True, 

64 summary="Update current user profile", 

65 responses=parse_responses( 

66 { 

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

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

69 }, 

70 common_responses, 

71 ), 

72) 

73async def update_user_profile_api( 

74 user_update: UserUpdate, 

75 response: Response, 

76 token: dict = Depends(verify_token), 

77 db: AsyncSession = Depends(get_db), 

78 redis_client=Depends(get_redis), 

79 mailer: SMTPMailer = Depends(get_mailer), 

80): 

81 """ 

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

83 """ 

84 try: 

85 user_id = token.get("sub") 

86 result = await update_user_profile(db, user_id, user_update, mailer, redis_client) 

87 

88 if not result: 

89 raise NotFoundException("User not found") 

90 

91 user, email_change_requested = result 

92 

93 user_data = _to_user_profile(user) 

94 

95 if email_change_requested: 

96 response.status_code = 202 

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

98 

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

100 except NotFoundException: 

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

102 except ValueError: 

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

104 except Exception: 

105 raise HTTPException(status_code=500) 

106 

107 

108@router.put( 

109 "/password", 

110 response_model=APIResponse[None], 

111 response_model_exclude_unset=True, 

112 summary="Change current user password", 

113 responses=parse_responses( 

114 { 

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

116 401: ( 

117 "Unauthorized", 

118 None, 

119 make_error_examples( 

120 401, 

121 { 

122 "invalidToken": "Invalid or expired token", 

123 "incorrectPassword": "Current password is incorrect", 

124 }, 

125 ), 

126 ), 

127 }, 

128 common_responses, 

129 ), 

130) 

131async def change_user_password_api( 

132 password_change: PasswordChange, 

133 token: dict = Depends(verify_token), 

134 db: AsyncSession = Depends(get_db), 

135 redis_client=Depends(get_redis), 

136): 

137 """ 

138 Change the current authenticated user's password. 

139 """ 

140 try: 

141 user_id = token.get("sub") 

142 current_session_id = token.get("sid") 

143 success = await change_password( 

144 db, 

145 user_id, 

146 password_change, 

147 redis_client, 

148 current_session_id=current_session_id, 

149 ) 

150 

151 if success: 

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

153 

154 except AuthenticationException: 

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

156 except Exception: 

157 raise HTTPException(status_code=500)