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

1import logging 

2from pathlib import Path 

3from typing import Any 

4 

5import yaml 

6from jinja2 import Environment, FileSystemLoader 

7from sqlalchemy.orm.session import Session 

8 

9from couchers.config import config 

10from couchers.jobs.enqueue import queue_job 

11from couchers.metrics import emails_counter 

12from couchers.proto.internal import jobs_pb2 

13 

14logger = logging.getLogger(__name__) 

15 

16loader = FileSystemLoader(Path(__file__).parent / ".." / ".." / ".." / "templates") 

17env = Environment(loader=loader, trim_blocks=True) 

18 

19 

20def _queue_email( 

21 session: Session, 

22 sender_name: str, 

23 sender_email: str, 

24 recipient: str, 

25 subject: str, 

26 plain: str, 

27 html: str | None, 

28 list_unsubscribe_header: str | None, 

29 source_data: str | None, 

30) -> None: 

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 

52def queue_email( 

53 session: Session, 

54 sender_name: str, 

55 sender_email: str, 

56 recipient: str, 

57 subject: str, 

58 plain: str, 

59 html: str | None, 

60 list_unsubscribe_header: str | None = None, 

61 source_data: str | None = None, 

62) -> None: 

63 """ 

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

65 """ 

66 _queue_email( 

67 session=session, 

68 sender_name=sender_name, 

69 sender_email=sender_email, 

70 recipient=recipient, 

71 subject=subject, 

72 plain=plain, 

73 html=html, 

74 list_unsubscribe_header=list_unsubscribe_header, 

75 source_data=source_data, 

76 ) 

77 

78 

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

80 source, _, _ = loader.get_source(env, f"system/{template_name}.md") 

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

82 

83 rendered_frontmatter = env.from_string(frontmatter_source).render(**template_args, plain=True, html=False) 

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

85 

86 plain = env.from_string(text_source.strip()).render( 

87 {**template_args, "frontmatter": frontmatter}, plain=True, html=False 

88 ) 

89 

90 queue_email( 

91 session, 

92 sender_name=config["NOTIFICATION_EMAIL_SENDER"], 

93 sender_email=config["NOTIFICATION_EMAIL_ADDRESS"], 

94 recipient=recipient, 

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

96 plain=plain, 

97 html=None, 

98 source_data=template_name, 

99 ) 

100 

101 emails_counter.inc()