Coverage for app / backend / src / couchers / email / queuing.py: 100%

1from pathlib import Path 

2from typing import Any 

3 

4import yaml 

5from sqlalchemy.orm.session import Session 

6 

7from couchers.config import config 

8from couchers.i18n import LocalizationContext 

9from couchers.jobs.enqueue import queue_job 

10from couchers.metrics import emails_counter 

11from couchers.proto.internal import jobs_pb2 

12from couchers.templating import Jinja2Template, template_folder 

13from couchers.utils import now 

14 

15 

16def _queue_email( 

17 session: Session, 

18 sender_name: str, 

19 sender_email: str, 

20 recipient: str, 

21 subject: str, 

22 plain: str, 

23 html: str | None, 

24 list_unsubscribe_header: str | None, 

25 source_data: str | None, 

26) -> None: 

27 """ 

28 This indirection is so that this can be easily mocked. Not sure how to do it better :( 

29 """ 

30 

31 # Import here to avoid circular dependency 

32 from couchers.jobs.handlers import send_email 

33 

34 payload = jobs_pb2.SendEmailPayload( 

35 sender_name=sender_name, 

36 sender_email=sender_email, 

37 recipient=recipient, 

38 subject=subject, 

39 plain=plain, 

40 html=html, 

41 list_unsubscribe_header=list_unsubscribe_header, 

42 source_data=source_data, 

43 ) 

44 queue_job( 

45 session, 

46 job=send_email, 

47 payload=payload, 

48 priority=5, 

49 ) 

50 

51 emails_counter.inc() 

52 

53 

54def queue_email( 

55 session: Session, 

56 sender_name: str, 

57 sender_email: str, 

58 recipient: str, 

59 subject: str, 

60 plain: str, 

61 html: str | None, 

62 list_unsubscribe_header: str | None = None, 

63 source_data: str | None = None, 

64) -> None: 

65 _queue_email( 

66 session=session, 

67 sender_name=sender_name, 

68 sender_email=sender_email, 

69 recipient=recipient, 

70 subject=subject, 

71 plain=plain, 

72 html=html, 

73 list_unsubscribe_header=list_unsubscribe_header, 

74 source_data=source_data, 

75 ) 

76 

77 

78def queue_userless_email( 

79 session: Session, recipient: str, subject: str, template_name: str, template_args: dict[str, Any] 

80) -> None: 

81 """ 

82 This is a simplified version of couchers.notifications.background._send_email_notification 

83 

84 It's for the few security emails where we don't have a user to email but send directly to an email address. 

85 """ 

86 

87 # Not yet localizable 

88 loc_context = LocalizationContext.en_utc() 

89 

90 template_args = { 

91 **template_args, 

92 "header_subject": subject, 

93 "footer_timezone_name": loc_context.localized_timezone, 

94 "footer_copyright_year": now().year, 

95 "footer_email_is_critical": True, # Results in no unsubscribe footer. 

96 } 

97 

98 # Format plaintext template 

99 plain_tmplt_body = (template_folder / f"{template_name}.txt").read_text() 

100 plain_tmplt_footer = (template_folder / "_footer.txt").read_text() 

101 plain_tmplt = Jinja2Template(source=plain_tmplt_body + plain_tmplt_footer, html=False) 

102 plain = plain_tmplt.render(template_args, loc_context) 

103 

104 # Format html template 

105 html_tmplt = Jinja2Template( 

106 source=(template_folder / "generated_html" / f"{template_name}.html").read_text(), html=True 

107 ) 

108 html = html_tmplt.render(template_args, loc_context) 

109 

110 queue_email( 

111 session, 

112 sender_name=config["NOTIFICATION_EMAIL_SENDER"], 

113 sender_email=config["NOTIFICATION_EMAIL_ADDRESS"], 

114 recipient=recipient, 

115 subject=config["NOTIFICATION_PREFIX"] + subject, 

116 plain=plain, 

117 html=html, 

118 source_data=config["VERSION"] + f"/{template_name}", 

119 ) 

120 

121 

122_system_email_templates_dir = Path(__file__).parent / ".." / ".." / ".." / "templates" / "system" 

123 

124 

125def queue_system_email(session: Session, recipient: str, template_name: str, template_args: dict[str, Any]) -> None: 

126 source = (_system_email_templates_dir / f"{template_name}.md").read_text(encoding="utf8") 

127 _, frontmatter_source, text_source = source.split("---", 2) 

128 

129 loc_context = LocalizationContext.en_utc() 

130 rendered_frontmatter = Jinja2Template(source=frontmatter_source, html=False).render(template_args, loc_context) 

131 frontmatter = yaml.load(rendered_frontmatter, Loader=yaml.FullLoader) 

132 

133 plain = Jinja2Template(source=text_source.strip(), html=False).render(template_args, loc_context) 

134 

135 queue_email( 

136 session, 

137 sender_name=config["NOTIFICATION_EMAIL_SENDER"], 

138 sender_email=config["NOTIFICATION_EMAIL_ADDRESS"], 

139 recipient=recipient, 

140 subject=config["NOTIFICATION_PREFIX"] + frontmatter["subject"], 

141 plain=plain, 

142 html=None, 

143 source_data=template_name, 

144 )