Coverage for polar/email/sender.py: 65%

1from abc import ABC, abstractmethod 2 ctx1ab

2from collections.abc import Iterable 2 ctx1ab

3from typing import Any, TypedDict 2 ctx1ab

4 

5import httpx 2 ctx1ab

6import structlog 2 ctx1ab

7from email_validator import validate_email 2 ctx1ab

8 

9from polar.config import EmailSender as EmailSenderType 2 ctx1ab

10from polar.config import settings 2 ctx1ab

11from polar.exceptions import PolarError 2 ctx1ab

12from polar.logging import Logger 2 ctx1ab

13from polar.worker import enqueue_job 2 ctx1ab

14 

15log: Logger = structlog.get_logger() 2 ctx1ab

16 

17DEFAULT_FROM_NAME = settings.EMAIL_FROM_NAME 2 ctx1ab

18DEFAULT_FROM_EMAIL_ADDRESS = f"{settings.EMAIL_FROM_LOCAL}@{settings.EMAIL_FROM_DOMAIN}" 2 ctx1ab

19DEFAULT_REPLY_TO_NAME = "Polar Support" 2 ctx1ab

20DEFAULT_REPLY_TO_EMAIL_ADDRESS = "support@polar.sh" 2 ctx1ab

21 

22 

23def to_ascii_email(email: str) -> str: 2 ctx1ab

24 """ 

25 Convert an email address to ASCII format, possibly using punycode for internationalized domains. 

26 """ 

27 validated_email = validate_email(email, check_deliverability=False) 

28 return validated_email.ascii_email or email 

29 

30 

31class EmailSenderError(PolarError): ... 2 ctx1ab

32 

33 

34class SendEmailError(EmailSenderError): 2 ctx1ab

35 def __init__(self, message: str) -> None: 2 ctx1ab

36 super().__init__(message) 

37 

38 

39class Attachment(TypedDict): 2 ctx1ab

40 remote_url: str 2 ctx1ab

41 filename: str 2 ctx1ab

42 

43 

44class EmailSender(ABC): 2 ctx1ab

45 @abstractmethod 2 ctx1ab

46 async def send( 2 ctx1ab

47 self, 

48 *, 

49 to_email_addr: str, 

50 subject: str, 

51 html_content: str, 

52 from_name: str = DEFAULT_FROM_NAME, 

53 from_email_addr: str = DEFAULT_FROM_EMAIL_ADDRESS, 

54 email_headers: dict[str, str] | None = None, 

55 reply_to_name: str | None = DEFAULT_REPLY_TO_NAME, 

56 reply_to_email_addr: str | None = DEFAULT_REPLY_TO_EMAIL_ADDRESS, 

57 attachments: Iterable[Attachment] | None = None, 

58 ) -> None: 

59 pass 

60 

61 

62class LoggingEmailSender(EmailSender): 2 ctx1ab

63 async def send( 2 ctx1ab

64 self, 

65 *, 

66 to_email_addr: str, 

67 subject: str, 

68 html_content: str, 

69 from_name: str = DEFAULT_FROM_NAME, 

70 from_email_addr: str = DEFAULT_FROM_EMAIL_ADDRESS, 

71 email_headers: dict[str, str] | None = None, 

72 reply_to_name: str | None = DEFAULT_REPLY_TO_NAME, 

73 reply_to_email_addr: str | None = DEFAULT_REPLY_TO_EMAIL_ADDRESS, 

74 attachments: Iterable[Attachment] | None = None, 

75 ) -> None: 

76 log.info( 

77 "Sending an email", 

78 to_email_addr=to_ascii_email(to_email_addr), 

79 subject=subject, 

80 from_name=from_name, 

81 from_email_addr=to_ascii_email(from_email_addr), 

82 ) 

83 

84 

85class ResendEmailSender(EmailSender): 2 ctx1ab

86 def __init__(self) -> None: 2 ctx1ab

87 self.client = httpx.AsyncClient( 

88 base_url=settings.RESEND_API_BASE_URL, 

89 headers={"Authorization": f"Bearer {settings.RESEND_API_KEY}"}, 

90 ) 

91 

92 async def send( 2 ctx1ab

93 self, 

94 *, 

95 to_email_addr: str, 

96 subject: str, 

97 html_content: str, 

98 from_name: str = DEFAULT_FROM_NAME, 

99 from_email_addr: str = DEFAULT_FROM_EMAIL_ADDRESS, 

100 email_headers: dict[str, str] | None = None, 

101 reply_to_name: str | None = DEFAULT_REPLY_TO_NAME, 

102 reply_to_email_addr: str | None = DEFAULT_REPLY_TO_EMAIL_ADDRESS, 

103 attachments: Iterable[Attachment] | None = None, 

104 ) -> None: 

105 to_email_addr_ascii = to_ascii_email(to_email_addr) 

106 payload: dict[str, Any] = { 

107 "from": f"{from_name} <{to_ascii_email(from_email_addr)}>", 

108 "to": [to_email_addr_ascii], 

109 "subject": subject, 

110 "html": html_content, 

111 "headers": email_headers or {}, 

112 "attachments": [ 

113 { 

114 "path": attachment["remote_url"], 

115 "filename": attachment["filename"], 

116 } 

117 for attachment in attachments 

118 ] 

119 if attachments 

120 else [], 

121 } 

122 if reply_to_name and reply_to_email_addr: 

123 payload["reply_to"] = ( 

124 f"{reply_to_name} <{to_ascii_email(reply_to_email_addr)}>" 

125 ) 

126 

127 try: 

128 response = await self.client.post("/emails", json=payload) 

129 response.raise_for_status() 

130 email = response.json() 

131 except httpx.HTTPError as e: 

132 log.warning( 

133 "resend.send_error", 

134 to_email_addr=to_email_addr_ascii, 

135 subject=subject, 

136 error=e, 

137 ) 

138 raise SendEmailError(str(e)) from e 

139 

140 log.info( 

141 "resend.send", 

142 to_email_addr=to_email_addr_ascii, 

143 subject=subject, 

144 email_id=email["id"], 

145 ) 

146 

147 

148class EmailFromReply(TypedDict): 2 ctx1ab

149 from_name: str 2 ctx1ab

150 from_email_addr: str 2 ctx1ab

151 reply_to_name: str 2 ctx1ab

152 reply_to_email_addr: str 2 ctx1ab

153 

154 

155def enqueue_email( 2 ctx1ab

156 to_email_addr: str, 

157 subject: str, 

158 html_content: str, 

159 from_name: str = DEFAULT_FROM_NAME, 

160 from_email_addr: str = DEFAULT_FROM_EMAIL_ADDRESS, 

161 email_headers: dict[str, str] | None = None, 

162 reply_to_name: str | None = DEFAULT_REPLY_TO_NAME, 

163 reply_to_email_addr: str | None = DEFAULT_REPLY_TO_EMAIL_ADDRESS, 

164 attachments: Iterable[Attachment] | None = None, 

165) -> None: 

166 enqueue_job( 

167 "email.send", 

168 to_email_addr=to_email_addr, 

169 subject=subject, 

170 html_content=html_content, 

171 from_name=from_name, 

172 from_email_addr=from_email_addr, 

173 email_headers=email_headers, 

174 reply_to_name=reply_to_name, 

175 reply_to_email_addr=reply_to_email_addr, 

176 attachments=attachments, 

177 ) 

178 

179 

180email_sender: EmailSender 2 ctx1ab

181if settings.EMAIL_SENDER == EmailSenderType.resend: 181 ↛ 182line 181 didn't jump to line 182 because the condition on line 181 was never true2 ctx1ab

182 email_sender = ResendEmailSender() 

183else: 

184 # Logging in development 

185 email_sender = LoggingEmailSender() 2 ctx1ab