Coverage for app/utils.py: 89%

1import logging (empty)

2from dataclasses import dataclass (empty)

3from datetime import UTC, datetime, timedelta (empty)

4from pathlib import Path (empty)

5from typing import Any (empty)

6 

7import emails (empty)

8import jwt (empty)

9from jinja2 import Template (empty)

10from jwt.exceptions import InvalidTokenError (empty)

11 

12from app.core import security (empty)

13from app.core.config import settings (empty)

14 

15logging.basicConfig(level=logging.INFO) (empty)

16logger = logging.getLogger(__name__) (empty)

17 

18 

19@dataclass (empty)

20class EmailData: (empty)

21 html_content: str 

22 subject: str 

23 

24 

25def render_email_template(*, template_name: str, context: dict[str, Any]) -> str: (empty)

26 template_str = ( 2 ctx1bd

27 Path(__file__).parent / "email-templates" / template_name 

28 ).read_text() 

29 template: Template = Template(template_str) 2 ctx1bd

30 html_content = template.render(context) (empty)

31 return html_content (empty)

32 

33 

34def send_email( (empty)

35 *, 

36 email_to: str, 

37 subject: str = "", 

38 html_content: str = "", 

39) -> None: 

40 assert settings.emails_enabled, "no provided configuration for email variables" (empty)

41 assert settings.EMAILS_FROM_EMAIL # For type checker (empty)

42 message = emails.message.Message( (empty)

43 subject=subject, 

44 html=html_content, 

45 mail_from=(settings.EMAILS_FROM_NAME, settings.EMAILS_FROM_EMAIL), 

46 ) 

47 smtp_options = {"host": settings.SMTP_HOST, "port": settings.SMTP_PORT} (empty)

48 if settings.SMTP_TLS: (empty)

49 smtp_options["tls"] = True 

50 elif settings.SMTP_SSL: (empty)

51 smtp_options["ssl"] = True 

52 if settings.SMTP_USER: (empty)

53 smtp_options["user"] = settings.SMTP_USER (empty)

54 if settings.SMTP_PASSWORD: (empty)

55 smtp_options["password"] = settings.SMTP_PASSWORD 

56 response = message.send(to=email_to, smtp=smtp_options) (empty)

57 logger.info(f"send email result: {response}") (empty)

58 

59 

60def generate_test_email(email_to: str) -> EmailData: (empty)

61 project_name = settings.PROJECT_NAME 

62 subject = f"{project_name} - Test email" 

63 html_content = render_email_template( 

64 template_name="test_email.html", 

65 context={"project_name": settings.PROJECT_NAME, "email": email_to}, 

66 ) 

67 return EmailData(html_content=html_content, subject=subject) 

68 

69 

70def generate_reset_password_email(email_to: str, email: str, token: str) -> EmailData: (empty)

71 project_name = settings.PROJECT_NAME 1 ctx1b

72 subject = f"{project_name} - Password recovery for user {email}" 1 ctx1b

73 link = f"{settings.FRONTEND_HOST}/reset-password?token={token}" 1 ctx1b

74 html_content = render_email_template( 1 ctx1b

75 template_name="reset_password.html", 

76 context={ 

77 "project_name": settings.PROJECT_NAME, 

78 "username": email, 

79 "email": email_to, 

80 "valid_hours": settings.EMAIL_RESET_TOKEN_EXPIRE_HOURS, 

81 "link": link, 

82 }, 

83 ) 

84 return EmailData(html_content=html_content, subject=subject) (empty)

85 

86 

87def generate_new_account_email( (empty)

88 email_to: str, username: str, password: str 

89) -> EmailData: 

90 project_name = settings.PROJECT_NAME 1 ctx1d

91 subject = f"{project_name} - New account for user {username}" 1 ctx1d

92 html_content = render_email_template( 1 ctx1d

93 template_name="new_account.html", 

94 context={ 

95 "project_name": settings.PROJECT_NAME, 

96 "username": username, 

97 "password": password, 

98 "email": email_to, 

99 "link": settings.FRONTEND_HOST, 

100 }, 

101 ) 

102 return EmailData(html_content=html_content, subject=subject) (empty)

103 

104 

105def generate_password_reset_token(email: str) -> str: (empty)

106 delta = timedelta(hours=settings.EMAIL_RESET_TOKEN_EXPIRE_HOURS) 2 ctx1bc

107 now = datetime.now(UTC) 2 ctx1bc

108 expires = now + delta 2 ctx1bc

109 exp = expires.timestamp() 2 ctx1bc

110 encoded_jwt = jwt.encode( 2 ctx1bc

111 {"exp": exp, "nbf": now, "sub": email}, 

112 settings.SECRET_KEY, 

113 algorithm=security.ALGORITHM, 

114 ) 

115 return encoded_jwt 2 ctx1bc

116 

117 

118def verify_password_reset_token(token: str) -> str | None: (empty)

119 try: 2 ctx1ce

120 decoded_token = jwt.decode( 2 ctx1ce

121 token, settings.SECRET_KEY, algorithms=[security.ALGORITHM] 

122 ) 

123 return str(decoded_token["sub"]) 1 ctx1c

124 except InvalidTokenError: 1 ctx1e

125 return None 1 ctx1e