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

1import logging 

2from urllib.parse import quote 

3 

4import redis 

5from sqlalchemy import or_, select 

6from sqlalchemy.ext.asyncio import AsyncSession 

7 

8from api.auth.services import _request_email_change_verification_email 

9from core.config import settings 

10from core.security import clear_user_all_sessions, hash_password, verify_password 

11from extensions.smtp import SMTPMailer 

12from models.users import Users 

13from utils.custom_exception import ( 

14 AuthenticationException, 

15 ServerException, 

16 SMTPNotConfiguredException, 

17) 

18from utils.email_templates import EMAIL_VERIFICATION_TEMPLATE 

19 

20from .schema import PasswordChange, UserUpdate 

21 

22logger = logging.getLogger("account") 

23 

24 

25async def get_user_by_id(db: AsyncSession, user_id: str) -> Users | None: 

26 """Get user info by id""" 

27 result = await db.execute(select(Users).where(Users.id == user_id)) 

28 return result.scalar_one_or_none() 

29 

30 

31async def update_user_profile( 

32 db: AsyncSession, 

33 user_id: str, 

34 user_update: UserUpdate, 

35 mailer: SMTPMailer | None = None, 

36 redis_client: redis.Redis | None = None, 

37) -> tuple[Users, bool] | None: 

38 """Update user info (excluding password)""" 

39 user = await get_user_by_id(db, user_id) 

40 if not user: 

41 return None 

42 

43 email_change_requested = False 

44 new_email = user_update.email 

45 if new_email and new_email != user.email: 

46 result = await db.execute( 

47 select(Users).where( 

48 or_(Users.email == new_email, Users.pending_email == new_email), Users.id != user_id 

49 ) 

50 ) 

51 if result.scalar_one_or_none(): 

52 raise ValueError("Email already exists") 

53 

54 # Defer email change until verification completes. 

55 user.pending_email = new_email 

56 email_change_requested = True 

57 

58 update_data = user_update.model_dump(exclude_unset=True) 

59 update_data.pop("email", None) 

60 for field, value in update_data.items(): 

61 setattr(user, field, value) 

62 

63 if ( 

64 email_change_requested 

65 and settings.SMTP_ENABLE 

66 and mailer 

67 and getattr(mailer, "enabled", False) 

68 ): 

69 should_send = True 

70 if redis_client: 

71 cooldown_key = f"email_verification_cooldown:{new_email}" 

72 remaining_seconds = await redis_client.ttl(cooldown_key) 

73 try: 

74 remaining_seconds = int(remaining_seconds) 

75 except TypeError, ValueError: 

76 remaining_seconds = 0 

77 if remaining_seconds > 0: 

78 should_send = False 

79 

80 if should_send: 

81 try: 

82 token_meta = await _request_email_change_verification_email(db, user, new_email) 

83 verification_url = ( 

84 f"http{'s' if settings.SSL_ENABLE else ''}://" 

85 f"{settings.HOSTNAME}:{settings.FRONTEND_PORT}" 

86 f"/auth/verify-email?token={quote(token_meta['verification_token'], safe='')}" 

87 ) 

88 

89 user_name = f"{user.first_name} {user.last_name}".strip() 

90 app_name = settings.PROJECT_NAME 

91 

92 email_content = EMAIL_VERIFICATION_TEMPLATE.render( 

93 verification_url=verification_url, 

94 user_name=user_name, 

95 app_name=app_name, 

96 expire_minutes=settings.EMAIL_VERIFICATION_TOKEN_EXPIRE_MINUTES, 

97 ) 

98 

99 mailer.send_text( 

100 to_emails=[new_email], 

101 subject=email_content["subject"], 

102 body=email_content["body"], 

103 html_body=email_content.get("html_body"), 

104 ) 

105 

106 if redis_client: 

107 await redis_client.setex( 

108 cooldown_key, settings.EMAIL_VERIFICATION_COOLDOWN_SECONDS, "1" 

109 ) 

110 except SMTPNotConfiguredException as exc: 

111 logger.warning("Skip email change verification send: %s", exc) 

112 

113 await db.commit() 

114 await db.refresh(user) 

115 return user, email_change_requested 

116 

117 

118async def change_password( 

119 db: AsyncSession, 

120 user_id: str, 

121 password_change: PasswordChange, 

122 redis_client=None, 

123 current_session_id: str | None = None, 

124) -> bool: 

125 """Change user password""" 

126 try: 

127 user = await get_user_by_id(db, user_id) 

128 if not user: 

129 return False 

130 

131 if not await verify_password(password_change.current_password, user.hash_password): 

132 raise AuthenticationException("Current password is incorrect") 

133 

134 user.hash_password = await hash_password(password_change.new_password) 

135 user.password_reset_required = False 

136 

137 if password_change.logout_other_devices and redis_client: 

138 await clear_user_all_sessions( 

139 db, 

140 redis_client, 

141 user_id, 

142 exclude_session_id=current_session_id, 

143 ) 

144 

145 await db.commit() 

146 

147 return True 

148 except AuthenticationException: 

149 raise 

150 except Exception as e: 

151 raise ServerException(f"Failed to change password: {e}")