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(session: Session, payload: jobs_pb2.SendEmailPayload) -> None: 

17 """ 

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

19 """ 

20 

21 # Import here to avoid circular dependency 

22 from couchers.jobs.handlers import send_email # noqa: PLC0415 

23 

24 queue_job( 

25 session, 

26 job=send_email, 

27 payload=payload, 

28 priority=5, 

29 ) 

30 

31 emails_counter.inc() 

32 

33 

34def queue_email(session: Session, payload: jobs_pb2.SendEmailPayload) -> None: 

35 _queue_email(session, payload) 

36 

37 

38def queue_userless_email( 

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

40) -> None: 

41 """ 

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

43 

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

45 """ 

46 

47 # Not yet localizable 

48 loc_context = LocalizationContext.en_utc() 

49 

50 template_args = { 

51 **template_args, 

52 "header_subject": subject, 

53 "footer_timezone_name": loc_context.localized_timezone, 

54 "footer_copyright_year": now().year, 

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

56 } 

57 

58 # Format plaintext template 

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

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

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

62 plain = plain_tmplt.render(template_args, loc_context) 

63 

64 # Format html template 

65 html_tmplt = Jinja2Template( 

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

67 ) 

68 html = html_tmplt.render(template_args, loc_context) 

69 

70 queue_email( 

71 session, 

72 jobs_pb2.SendEmailPayload( 

73 sender_name=config["NOTIFICATION_EMAIL_SENDER"], 

74 sender_email=config["NOTIFICATION_EMAIL_ADDRESS"], 

75 recipient=recipient, 

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

77 plain=plain, 

78 html=html, 

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

80 ), 

81 ) 

82 

83 

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

85 

86 

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

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

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

90 

91 loc_context = LocalizationContext.en_utc() 

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

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

94 

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

96 

97 queue_email( 

98 session, 

99 jobs_pb2.SendEmailPayload( 

100 sender_name=config["NOTIFICATION_EMAIL_SENDER"], 

101 sender_email=config["NOTIFICATION_EMAIL_ADDRESS"], 

102 recipient=recipient, 

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

104 plain=plain, 

105 source_data=template_name, 

106 ), 

107 )