Coverage for 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 payload = jobs_pb2.SendEmailPayload( 

32 sender_name=sender_name, 

33 sender_email=sender_email, 

34 recipient=recipient, 

35 subject=subject, 

36 plain=plain, 

37 html=html, 

38 list_unsubscribe_header=list_unsubscribe_header, 

39 source_data=source_data, 

40 ) 

41 queue_job( 

42 session, 

43 job_type="send_email", 

44 payload=payload, 

45 priority=5, 

46 ) 

47 

48 

49def queue_email( 

50 session: Session, 

51 sender_name: str, 

52 sender_email: str, 

53 recipient: str, 

54 subject: str, 

55 plain: str, 

56 html: str | None, 

57 list_unsubscribe_header: str | None = None, 

58 source_data: str | None = None, 

59) -> None: 

60 """ 

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

62 """ 

63 _queue_email( 

64 session=session, 

65 sender_name=sender_name, 

66 sender_email=sender_email, 

67 recipient=recipient, 

68 subject=subject, 

69 plain=plain, 

70 html=html, 

71 list_unsubscribe_header=list_unsubscribe_header, 

72 source_data=source_data, 

73 ) 

74 

75 

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

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

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

79 

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

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

82 

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

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

85 ) 

86 

87 queue_email( 

88 session, 

89 sender_name=config["NOTIFICATION_EMAIL_SENDER"], 

90 sender_email=config["NOTIFICATION_EMAIL_ADDRESS"], 

91 recipient=recipient, 

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

93 plain=plain, 

94 html=None, 

95 source_data=template_name, 

96 ) 

97 

98 emails_counter.inc()