Coverage for app/backend/src/couchers/email/emails.py: 95%

1089 statements  

« prev     ^ index     » next       coverage.py v7.14.3, created at 2026-06-29 12:51 +0000

1""" 

2Defines data models for each email we sent out to users. 

3""" 

4 

5import re 

6from dataclasses import dataclass, replace 

7from datetime import UTC, date, datetime 

8from typing import Self, assert_never 

9 

10from markupsafe import Markup, escape 

11 

12from couchers import urls 

13from couchers.config import config 

14from couchers.constants import LATEST_RELEASE_BLOG_URL 

15from couchers.email.blocks import ( 

16 ActionBlock, 

17 EmailBase, 

18 EmailBlock, 

19 ParaBlock, 

20 QuoteBlock, 

21 UserInfo, 

22) 

23from couchers.email.locales import get_emails_i18next 

24from couchers.i18n import LocalizationContext 

25from couchers.i18n.localize import format_phone_number 

26from couchers.markup import markdown_to_plaintext 

27from couchers.notifications.quick_links import generate_quick_decline_link 

28from couchers.proto import conversations_pb2, events_pb2, notification_data_pb2 

29from couchers.utils import now, to_aware_datetime 

30 

31# Common string keys 

32_do_not_reply_request_string_key = "generic.do_not_reply_request" 

33 

34# Specific email definitions 

35 

36 

37@dataclass(kw_only=True, slots=True) 

38class AccountDeletionStartedEmail(EmailBase): 

39 """Sent to a user to confirm their account deletion request.""" 

40 

41 deletion_link: str 

42 

43 @property 

44 def string_key_base(self) -> str: 

45 return "account_deletion.started" 

46 

47 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

48 builder = self._body_builder(loc_context, security_warning=True) 

49 builder.para(".request_description") 

50 builder.para(".confirmation_instructions") 

51 builder.action(self.deletion_link, ".confirm_action") 

52 return builder.build() 

53 

54 @classmethod 

55 def from_notification(cls, data: notification_data_pb2.AccountDeletionStart, *, user_name: str) -> Self: 

56 return cls( 

57 user_name=user_name, 

58 deletion_link=urls.delete_account_link(account_deletion_token=data.deletion_token), 

59 ) 

60 

61 @classmethod 

62 def test_instances(cls) -> list[Self]: 

63 return [ 

64 cls( 

65 user_name="Alice", 

66 deletion_link="https://couchers.org/delete-account?token=xxx", 

67 ) 

68 ] 

69 

70 

71@dataclass(kw_only=True, slots=True) 

72class AccountDeletionCompletedEmail(EmailBase): 

73 """Sent to a user after their account has been deleted.""" 

74 

75 undelete_link: str 

76 days: int 

77 

78 @property 

79 def string_key_base(self) -> str: 

80 return "account_deletion.completed" 

81 

82 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

83 builder = self._body_builder(loc_context, security_warning=True) 

84 builder.para(".confirmation") 

85 builder.para(".farewell") 

86 builder.para(".recovery_instructions_days", {"count": self.days}) 

87 builder.action(self.undelete_link, ".recover_action") 

88 return builder.build() 

89 

90 @classmethod 

91 def from_notification(cls, data: notification_data_pb2.AccountDeletionComplete, *, user_name: str) -> Self: 

92 return cls( 

93 user_name=user_name, 

94 undelete_link=urls.recover_account_link(account_undelete_token=data.undelete_token), 

95 days=data.undelete_days, 

96 ) 

97 

98 @classmethod 

99 def test_instances(cls) -> list[Self]: 

100 return [ 

101 cls( 

102 user_name="Alice", 

103 undelete_link="https://couchers.org/recover-account?token=xxx", 

104 days=30, 

105 ) 

106 ] 

107 

108 

109@dataclass(kw_only=True, slots=True) 

110class AccountDeletionRecoveredEmail(EmailBase): 

111 """Sent to a user after their account deletion has been cancelled.""" 

112 

113 @property 

114 def string_key_base(self) -> str: 

115 return "account_deletion.recovered" 

116 

117 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

118 builder = self._body_builder(loc_context, security_warning=True) 

119 builder.para(".confirmation") 

120 builder.para(".login_instructions") 

121 builder.action(urls.app_link(), ".login_action") 

122 builder.para(".redelete_instructions") 

123 return builder.build() 

124 

125 @classmethod 

126 def test_instances(cls) -> list[Self]: 

127 return [cls(user_name="Alice")] 

128 

129 

130@dataclass(kw_only=True, slots=True) 

131class ActivenessProbeEmail(EmailBase): 

132 """Sent to a host to check if they are still open to hosting.""" 

133 

134 days_left: int 

135 

136 @property 

137 def string_key_base(self) -> str: 

138 return "activeness_probe" 

139 

140 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

141 builder = self._body_builder(loc_context) 

142 builder.para(".body") 

143 builder.para(".instructions_days", {"count": self.days_left}) 

144 builder.action(urls.app_link(), ".login_action") 

145 builder.para(".encouragement") 

146 

147 # Extract major.minor from the version string. "v1.3.18927" -> "1.3" 

148 version = config.VERSION 

149 if version_match := re.search(r"^v?(\d+\.\d+)\b", version): 149 ↛ 150line 149 didn't jump to line 150 because the condition on line 149 was never true

150 version = version_match[1] 

151 

152 builder.para(".latest_release", {"version": version}) 

153 builder.action(LATEST_RELEASE_BLOG_URL, ".read_blog_action") 

154 return builder.build() 

155 

156 @classmethod 

157 def from_notification(cls, data: notification_data_pb2.ActivenessProbe, *, user_name: str) -> Self: 

158 days_left = (to_aware_datetime(data.deadline) - now()).days 

159 return cls(user_name=user_name, days_left=days_left) 

160 

161 @classmethod 

162 def test_instances(cls) -> list[Self]: 

163 return [cls(user_name="Alice", days_left=7)] 

164 

165 

166@dataclass(kw_only=True, slots=True) 

167class APIKeyIssuedEmail(EmailBase): 

168 """Sent to a user to notify them that their API key was issued.""" 

169 

170 api_key: str 

171 expiry: datetime 

172 

173 @property 

174 def string_key_base(self) -> str: 

175 return "api_key_issued" 

176 

177 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

178 builder = self._body_builder(loc_context, security_warning=True) 

179 builder.para(".header") 

180 builder.quote(self.api_key, markdown=False) 

181 builder.para(".expiry", {"datetime": loc_context.localize_datetime(self.expiry)}) 

182 builder.para(".usage_warning") 

183 builder.para(".policy_warning") 

184 return builder.build() 

185 

186 @classmethod 

187 def from_notification(cls, data: notification_data_pb2.ApiKeyCreate, *, user_name: str) -> Self: 

188 return cls(user_name=user_name, api_key=data.api_key, expiry=data.expiry.ToDatetime(tzinfo=UTC)) 

189 

190 @classmethod 

191 def test_instances(cls) -> list[Self]: 

192 return [cls(user_name="Alice", api_key="my_api_key_123", expiry=datetime(2099, 12, 31, 23, 59, 59, tzinfo=UTC))] 

193 

194 

195@dataclass(kw_only=True, slots=True) 

196class BadgeChangedEmail(EmailBase): 

197 """Sent to a user to notify them that a badge was added or removed from their profile.""" 

198 

199 badge_name: str 

200 added: bool 

201 

202 @property 

203 def string_key_base(self) -> str: 

204 return "badges.added" if self.added else "badges.removed" 

205 

206 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

207 return self._localize(loc_context, ".subject", {"name": self.badge_name}) 

208 

209 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

210 builder = self._body_builder(loc_context) 

211 builder.para(".body", {"name": self.badge_name}) 

212 return builder.build() 

213 

214 @classmethod 

215 def from_notification( 

216 cls, data: notification_data_pb2.BadgeAdd | notification_data_pb2.BadgeRemove, *, user_name: str 

217 ) -> Self: 

218 return cls( 

219 user_name=user_name, badge_name=data.badge_name, added=isinstance(data, notification_data_pb2.BadgeAdd) 

220 ) 

221 

222 @classmethod 

223 def test_instances(cls) -> list[Self]: 

224 prototype = cls(user_name="Alice", badge_name="Founder", added=True) 

225 return [replace(prototype, added=True), replace(prototype, added=False)] 

226 

227 

228@dataclass(kw_only=True, slots=True) 

229class BirthdateChangedEmail(EmailBase): 

230 """Sent to a user to notify them that their birthdate was changed.""" 

231 

232 new_birthdate: date 

233 

234 @property 

235 def string_key_base(self) -> str: 

236 return "birthdate_changed" 

237 

238 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

239 builder = self._body_builder(loc_context, security_warning=True) 

240 builder.para(".body", {"date": loc_context.localize_date(self.new_birthdate)}) 

241 return builder.build() 

242 

243 @classmethod 

244 def from_notification(cls, data: notification_data_pb2.BirthdateChange, *, user_name: str) -> Self: 

245 return cls(user_name=user_name, new_birthdate=date.fromisoformat(data.birthdate)) 

246 

247 @classmethod 

248 def test_instances(cls) -> list[Self]: 

249 return [ 

250 cls( 

251 user_name="Alice", 

252 new_birthdate=date(1990, 1, 1), 

253 ) 

254 ] 

255 

256 

257@dataclass(kw_only=True, slots=True) 

258class ChatMessageReceivedEmail(EmailBase): 

259 """Sent to a user when they receive a new chat message.""" 

260 

261 group_chat_title: str | None # None if direct message 

262 author: UserInfo 

263 text: str 

264 view_url: str 

265 

266 @property 

267 def string_key_base(self) -> str: 

268 return f"chat_messages.received.{'direct' if self.group_chat_title is None else 'group'}" 

269 

270 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

271 return self._localize( 

272 loc_context, ".subject", {"author": self.author.name, "group": self.group_chat_title or ""} 

273 ) 

274 

275 def get_preview_line(self, loc_context: LocalizationContext) -> str | None: 

276 return self.text 

277 

278 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

279 builder = self._body_builder(loc_context) 

280 builder.para(".body", {"author": self.author.name, "group": self.group_chat_title or ""}) 

281 builder.user(self.author) 

282 builder.quote(self.text, markdown=False) 

283 builder.action(self.view_url, ".view_action") 

284 return builder.build() 

285 

286 @classmethod 

287 def from_notification(cls, data: notification_data_pb2.ChatMessage, *, user_name: str) -> Self: 

288 return cls( 

289 user_name, 

290 author=UserInfo.from_protobuf(data.author), 

291 text=data.text, 

292 group_chat_title=data.group_chat_title or None, 

293 view_url=urls.chat_link(chat_id=data.group_chat_id), 

294 ) 

295 

296 @classmethod 

297 def test_instances(cls) -> list[Self]: 

298 prototype = cls( 

299 user_name="Alice", 

300 group_chat_title=None, 

301 author=UserInfo.dummy_bob(), 

302 text="Hi Alice!", 

303 view_url="https://couchers.org/messages/chats/123", 

304 ) 

305 return [ 

306 replace(prototype, group_chat_title=None), 

307 replace(prototype, group_chat_title="Best friends"), 

308 ] 

309 

310 

311@dataclass(kw_only=True, slots=True) 

312class ChatMessagesMissedEmail(EmailBase): 

313 """Sent to a user after they've missed new chat messages.""" 

314 

315 @dataclass(kw_only=True, slots=True) 

316 class Entry: 

317 """Entry for each chat with missed messages.""" 

318 

319 group_chat_title: str | None # None if direct message 

320 missed_count: int 

321 latest_message_author: UserInfo 

322 latest_message_text: str 

323 view_url: str 

324 

325 entries: list[Entry] 

326 

327 @property 

328 def string_key_base(self) -> str: 

329 return "chat_messages.missed" 

330 

331 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

332 return self._localize(loc_context, ".subject") 

333 

334 def get_preview_line(self, loc_context: LocalizationContext) -> str | None: 

335 if len(self.entries) != 1: 

336 return None 

337 return self.entries[0].latest_message_text 

338 

339 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

340 builder = self._body_builder(loc_context) 

341 for entry in self.entries: 

342 if entry.group_chat_title is None: 

343 builder.para(".in_dm", {"count": entry.missed_count, "author": entry.latest_message_author.name}) 

344 else: 

345 builder.para(".in_group", {"count": entry.missed_count, "group": entry.group_chat_title}) 

346 builder.user(entry.latest_message_author) 

347 builder.quote(entry.latest_message_text, markdown=False) 

348 builder.action(entry.view_url, ".view_action") 

349 return builder.build() 

350 

351 @classmethod 

352 def from_notification(cls, data: notification_data_pb2.ChatMissedMessages, *, user_name: str) -> Self: 

353 missed_entries = [ 

354 cls.Entry( 

355 group_chat_title=message.group_chat_title or None, 

356 missed_count=message.unseen_count, 

357 latest_message_author=UserInfo.from_protobuf(message.author), 

358 latest_message_text=message.text, 

359 view_url=urls.chat_link(chat_id=message.group_chat_id), 

360 ) 

361 for message in data.messages 

362 ] 

363 

364 return cls(user_name, entries=missed_entries) 

365 

366 @classmethod 

367 def test_instances(cls) -> list[Self]: 

368 entry_prototype = ChatMessagesMissedEmail.Entry( 

369 group_chat_title=None, 

370 missed_count=1, 

371 latest_message_author=UserInfo.dummy_bob(), 

372 latest_message_text="Hello!", 

373 view_url="https://couchers.org/messages/chats/123", 

374 ) 

375 return [ 

376 cls( 

377 user_name="Alice", 

378 entries=[ 

379 replace(entry_prototype, group_chat_title=None), 

380 replace(entry_prototype, group_chat_title="Best friends"), 

381 ], 

382 ) 

383 ] 

384 

385 

386@dataclass(kw_only=True, slots=True) 

387class DiscussionCreatedEmail(EmailBase): 

388 """Sent to a user when a new discussion is created in a community they follow.""" 

389 

390 author: UserInfo 

391 title: str 

392 parent_context: str # Community or group name 

393 markdown_text: str 

394 view_link: str 

395 

396 @property 

397 def string_key_base(self) -> str: 

398 return "discussions.created" 

399 

400 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

401 return self._localize(loc_context, ".subject", {"author": self.author.name, "title": self.title}) 

402 

403 def get_preview_line(self, loc_context: LocalizationContext) -> str | None: 

404 return markdown_to_plaintext(self.markdown_text) 

405 

406 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

407 builder = self._body_builder(loc_context) 

408 builder.para( 

409 ".body", 

410 { 

411 "author": self.author.name, 

412 "title": self.title, 

413 "parent_context": self.parent_context, 

414 }, 

415 ) 

416 builder.user(self.author) 

417 builder.quote(self.markdown_text, markdown=True) 

418 builder.action(self.view_link, ".view_action") 

419 return builder.build() 

420 

421 @classmethod 

422 def from_notification(cls, data: notification_data_pb2.DiscussionCreate, *, user_name: str) -> Self: 

423 discussion = data.discussion 

424 return cls( 

425 user_name=user_name, 

426 author=UserInfo.from_protobuf(data.author), 

427 title=discussion.title, 

428 parent_context=discussion.owner_title, 

429 markdown_text=discussion.content, 

430 view_link=urls.discussion_link(discussion_id=discussion.discussion_id, slug=discussion.slug), 

431 ) 

432 

433 @classmethod 

434 def test_instances(cls) -> list[Self]: 

435 return [ 

436 cls( 

437 user_name="Alice", 

438 author=UserInfo.dummy_bob(), 

439 title="Best hiking trails near Berlin", 

440 parent_context="Berlin", 

441 markdown_text="I've been exploring the area and found some **great** spots...", 

442 view_link="https://couchers.org/discussions/123", 

443 ) 

444 ] 

445 

446 

447@dataclass(kw_only=True, slots=True) 

448class DiscussionCommentEmail(EmailBase): 

449 """Sent to a user when someone comments on a discussion they follow.""" 

450 

451 author: UserInfo 

452 discussion_title: str 

453 discussion_parent_context: str # Community or group name 

454 markdown_text: str 

455 view_link: str 

456 

457 @property 

458 def string_key_base(self) -> str: 

459 return "discussions.comment" 

460 

461 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

462 return self._localize( 

463 loc_context, ".subject", {"author": self.author.name, "discussion_title": self.discussion_title} 

464 ) 

465 

466 def get_preview_line(self, loc_context: LocalizationContext) -> str | None: 

467 return markdown_to_plaintext(self.markdown_text) 

468 

469 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

470 builder = self._body_builder(loc_context) 

471 builder.para( 

472 ".body", 

473 { 

474 "author": self.author.name, 

475 "discussion_title": self.discussion_title, 

476 "parent_context": self.discussion_parent_context, 

477 }, 

478 ) 

479 builder.user(self.author) 

480 builder.quote(self.markdown_text, markdown=True) 

481 builder.action(self.view_link, ".view_action") 

482 return builder.build() 

483 

484 @classmethod 

485 def from_notification(cls, data: notification_data_pb2.DiscussionComment, *, user_name: str) -> Self: 

486 discussion = data.discussion 

487 return cls( 

488 user_name=user_name, 

489 author=UserInfo.from_protobuf(data.author), 

490 discussion_title=discussion.title, 

491 discussion_parent_context=discussion.owner_title, 

492 markdown_text=data.reply.content, 

493 view_link=urls.discussion_link(discussion_id=discussion.discussion_id, slug=discussion.slug), 

494 ) 

495 

496 @classmethod 

497 def test_instances(cls) -> list[Self]: 

498 return [ 

499 cls( 

500 user_name="Alice", 

501 author=UserInfo.dummy_bob(), 

502 discussion_title="Best hiking trails near Berlin", 

503 discussion_parent_context="Berlin", 

504 markdown_text="Great recommendations, I also **love** the Grünewald forest!", 

505 view_link="https://couchers.org/discussions/123", 

506 ) 

507 ] 

508 

509 

510@dataclass(kw_only=True, slots=True) 

511class DonationReceivedEmail(EmailBase): 

512 """Sent to a user to thank them for a donation.""" 

513 

514 amount: int 

515 receipt_url: str 

516 

517 @property 

518 def string_key_base(self) -> str: 

519 return "donation_received" 

520 

521 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

522 builder = self._body_builder(loc_context, standard_closing=False) 

523 builder.para(".thanks_amount", {"amount": self.amount}) 

524 builder.para(".purpose") 

525 builder.para(".invoice_receipt_info") 

526 builder.action(self.receipt_url, ".download_invoice") 

527 builder.para(".tax_acknowledgment") 

528 builder.para(".questions_contact") 

529 builder.para(".generosity_helps") 

530 builder.para(".thank_you") 

531 builder.para("generic.founders_signature") 

532 return builder.build() 

533 

534 @classmethod 

535 def from_notification(cls, data: notification_data_pb2.DonationReceived, *, user_name: str) -> Self: 

536 return cls(user_name=user_name, amount=data.amount, receipt_url=data.receipt_url) 

537 

538 @classmethod 

539 def test_instances(cls) -> list[Self]: 

540 return [cls(user_name="Alice", amount=25, receipt_url="https://couchers.org/receipts/123")] 

541 

542 

543@dataclass(kw_only=True, slots=True) 

544class EmailChangedEmail(EmailBase): 

545 """Sent to a user to notify them that their email address was changed.""" 

546 

547 new_email: str 

548 

549 @property 

550 def string_key_base(self) -> str: 

551 return "email_change.initiated" 

552 

553 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

554 builder = self._body_builder(loc_context, security_warning=True) 

555 builder.para(".body", {"email_address": self.new_email}) 

556 return builder.build() 

557 

558 @classmethod 

559 def from_notification(cls, data: notification_data_pb2.EmailAddressChange, *, user_name: str) -> Self: 

560 return cls(user_name=user_name, new_email=data.new_email) 

561 

562 @classmethod 

563 def test_instances(cls) -> list[Self]: 

564 return [cls(user_name="Alice", new_email="alice@example.com")] 

565 

566 

567@dataclass(kw_only=True, slots=True) 

568class EmailChangeConfirmationEmail(EmailBase): 

569 """Sent to a user to confirm their new email address.""" 

570 

571 old_email: str 

572 confirm_url: str 

573 

574 @property 

575 def string_key_base(self) -> str: 

576 return "email_change.confirmation" 

577 

578 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

579 builder = self._body_builder(loc_context, security_warning=True) 

580 builder.para(".context", {"old_email": self.old_email}) 

581 builder.para(".instructions") 

582 builder.action(self.confirm_url, ".confirm_action") 

583 return builder.build() 

584 

585 @classmethod 

586 def test_instances(cls) -> list[Self]: 

587 return [cls(user_name="Alice", old_email="alice@example.com", confirm_url="https://example.com")] 

588 

589 

590@dataclass(kw_only=True, slots=True) 

591class EmailVerifiedEmail(EmailBase): 

592 """Sent to a user to notify them that their new email address has been verified.""" 

593 

594 @property 

595 def string_key_base(self) -> str: 

596 return "email_change.verified" 

597 

598 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

599 builder = self._body_builder(loc_context, security_warning=True) 

600 builder.para(".body") 

601 return builder.build() 

602 

603 @classmethod 

604 def test_instances(cls) -> list[Self]: 

605 return [cls(user_name="Alice")] 

606 

607 

608@dataclass(kw_only=True, slots=True) 

609class EventInfo: 

610 """Common display fields for an event, extracted from its proto representation.""" 

611 

612 title: str 

613 start_time: datetime 

614 end_time: datetime 

615 online_link: str | None 

616 address: str | None 

617 view_url: str 

618 description_markdown: str 

619 

620 def get_details_block(self, loc_context: LocalizationContext) -> EmailBlock: 

621 # TODO(#8695): Support localized time ranges 

622 start_time_display = loc_context.localize_datetime(self.start_time, with_year=False, with_day_of_week=True) 

623 end_time_display = loc_context.localize_datetime(self.end_time, with_year=False, with_day_of_week=True) 

624 time_range_display = f"{start_time_display} - {end_time_display}" 

625 

626 # Format the following. Only "Online" is translated and it's on its own line, 

627 # so the string concatenation is fine. 

628 # **<title>** 

629 # <datetime-range> 

630 # *<address> / [Online](<online_link>)* 

631 html = f"<b>{escape(self.title)}</b>" 

632 html += "<br>" 

633 html += time_range_display 

634 if self.online_link: 634 ↛ 635line 634 didn't jump to line 635 because the condition on line 634 was never true

635 html += "<br>" 

636 online_link_text = loc_context.localize_string("events.generic.online_link", i18next=get_emails_i18next()) 

637 html += f'<i><a href="{escape(self.online_link)}">{escape(online_link_text)}</a></i>' 

638 elif self.address: 

639 html += "<br>" 

640 html += f"<i>{escape(self.address)}</i>" 

641 

642 return ParaBlock(text=Markup(html)) 

643 

644 def get_description_block(self) -> EmailBlock: 

645 return QuoteBlock(text=Markup(self.description_markdown), markdown=True) 

646 

647 def get_view_action_block(self, loc_context: LocalizationContext) -> EmailBlock: 

648 view_action_text = loc_context.localize_string("events.generic.view_action", i18next=get_emails_i18next()) 

649 return ActionBlock(text=view_action_text, target_url=self.view_url) 

650 

651 @classmethod 

652 def from_proto(cls, event: events_pb2.Event) -> EventInfo: 

653 return cls( 

654 title=event.title, 

655 start_time=event.start_time.ToDatetime(tzinfo=UTC), 

656 end_time=event.end_time.ToDatetime(tzinfo=UTC), 

657 online_link=event.online_information.link or None, 

658 address=event.offline_information.address or None, 

659 view_url=urls.event_link(occurrence_id=event.event_id, slug=event.slug), 

660 description_markdown=event.content or "", 

661 ) 

662 

663 @staticmethod 

664 def dummy() -> EventInfo: 

665 return EventInfo( 

666 title="Berlin Meetup", 

667 start_time=datetime(2025, 7, 15, 18, 0, 0, tzinfo=UTC), 

668 end_time=datetime(2025, 7, 15, 21, 0, 0, tzinfo=UTC), 

669 online_link=None, 

670 address="Alexanderplatz, Berlin", 

671 view_url="https://couchers.org/events/123/berlin-community-meetup", 

672 description_markdown="Come join us for our monthly meetup!", 

673 ) 

674 

675 

676@dataclass(kw_only=True, slots=True) 

677class EventCreatedEmail(EmailBase): 

678 """Sent when a user is invited to an event (create_approved) or a new event is created (create_any).""" 

679 

680 inviting_user: UserInfo 

681 event_info: EventInfo 

682 community_name: str | None 

683 community_url: str | None 

684 is_invite: bool # True = create_approved (invitation), False = create_any 

685 

686 @property 

687 def string_key_base(self) -> str: 

688 return f"events.created.{'invitation' if self.is_invite else 'notification'}" 

689 

690 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

691 return self._localize( 

692 loc_context, ".subject", {"user": self.inviting_user.name, "title": self.event_info.title} 

693 ) 

694 

695 def get_preview_line(self, loc_context: LocalizationContext) -> str | None: 

696 return markdown_to_plaintext(self.event_info.description_markdown) 

697 

698 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

699 builder = self._body_builder(loc_context) 

700 if self.community_name: 

701 builder.para(".body_with_community", {"community": self.community_name}) 

702 else: 

703 builder.para(".body_no_community") 

704 builder.block(self.event_info.get_details_block(loc_context)) 

705 builder.user(self.inviting_user) 

706 builder.block(self.event_info.get_description_block()) 

707 builder.block(self.event_info.get_view_action_block(loc_context)) 

708 return builder.build() 

709 

710 @classmethod 

711 def from_notification(cls, data: notification_data_pb2.EventCreate, *, user_name: str, is_invite: bool) -> Self: 

712 has_community = bool(data.in_community.community_id) 

713 community_url = ( 

714 urls.community_link(node_id=data.in_community.community_id, slug=data.in_community.slug) 

715 if has_community 

716 else None 

717 ) 

718 return cls( 

719 user_name=user_name, 

720 inviting_user=UserInfo.from_protobuf(data.inviting_user), 

721 event_info=EventInfo.from_proto(data.event), 

722 community_name=data.in_community.name if has_community else None, 

723 community_url=community_url, 

724 is_invite=is_invite, 

725 ) 

726 

727 @classmethod 

728 def test_instances(cls) -> list[Self]: 

729 prototype = cls( 

730 user_name="Alice", 

731 inviting_user=UserInfo.dummy_bob(), 

732 event_info=EventInfo.dummy(), 

733 community_name="Berlin", 

734 community_url="https://couchers.org/community/1/berlin-community", 

735 is_invite=True, 

736 ) 

737 return [ 

738 replace(prototype, is_invite=True), 

739 replace(prototype, is_invite=True, community_name=None, community_url=None), 

740 replace(prototype, is_invite=False), 

741 replace(prototype, is_invite=False, community_name=None, community_url=None), 

742 ] 

743 

744 

745@dataclass(kw_only=True, slots=True) 

746class EventUpdatedEmail(EmailBase): 

747 """Sent to subscribers when an event is updated.""" 

748 

749 updating_user: UserInfo 

750 event_info: EventInfo 

751 updated_items: list[notification_data_pb2.EventUpdateItem.ValueType] 

752 

753 @property 

754 def string_key_base(self) -> str: 

755 return "events.updated" 

756 

757 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

758 return self._localize( 

759 loc_context, ".subject", {"user": self.updating_user.name, "title": self.event_info.title} 

760 ) 

761 

762 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

763 builder = self._body_builder(loc_context) 

764 builder.para(".body") 

765 

766 updated_items_string_keys = list( 

767 filter(None, (type(self)._updated_item_to_string_key(i) for i in self.updated_items)) 

768 ) 

769 if updated_items_string_keys: 

770 updated_items_text = loc_context.localize_list( 

771 [self._localize(loc_context, key) for key in updated_items_string_keys] 

772 ) 

773 builder.para(".updated_items", {"items_list": updated_items_text}) 

774 else: 

775 builder.para(".updated_generic") 

776 

777 builder.block(self.event_info.get_details_block(loc_context)) 

778 builder.user(self.updating_user) 

779 builder.block(self.event_info.get_description_block()) 

780 builder.block(self.event_info.get_view_action_block(loc_context)) 

781 return builder.build() 

782 

783 @classmethod 

784 def from_notification(cls, data: notification_data_pb2.EventUpdate, *, user_name: str) -> Self: 

785 updated_items: list[notification_data_pb2.EventUpdateItem.ValueType] = [] 

786 if data.updated_enum_items: 786 ↛ 788line 786 didn't jump to line 788 because the condition on line 786 was always true

787 updated_items.extend(data.updated_enum_items) 

788 elif data.updated_str_items: 

789 for updated_str_item in data.updated_str_items: 

790 if updated_enum_item := cls._updated_item_str_to_enum(updated_str_item): 

791 updated_items.append(updated_enum_item) 

792 

793 return cls( 

794 user_name=user_name, 

795 updating_user=UserInfo.from_protobuf(data.updating_user), 

796 event_info=EventInfo.from_proto(data.event), 

797 updated_items=updated_items, 

798 ) 

799 

800 # TODO(#9117): Backcompat. Remove update_str_items fallback once known unused. 

801 @staticmethod 

802 def _updated_item_str_to_enum(value: str) -> notification_data_pb2.EventUpdateItem.ValueType | None: 

803 match value: 

804 case "title": 

805 return notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_TITLE 

806 case "content": 

807 return notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_CONTENT 

808 case "location": 

809 return notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_LOCATION 

810 case "start time": 

811 return notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_START_TIME 

812 case "end time": 

813 return notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_END_TIME 

814 case _: 

815 return None 

816 

817 @staticmethod 

818 def _updated_item_to_string_key(value: notification_data_pb2.EventUpdateItem.ValueType) -> str | None: 

819 match value: 

820 case notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_TITLE: 

821 return ".item_names.title" 

822 case notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_CONTENT: 

823 return ".item_names.content" 

824 case notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_LOCATION: 

825 return ".item_names.location" 

826 case notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_START_TIME: 

827 return ".item_names.start_time" 

828 case notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_END_TIME: 

829 return ".item_names.end_time" 

830 case _: 

831 return None 

832 

833 @classmethod 

834 def test_instances(cls) -> list[Self]: 

835 prototype = cls( 

836 user_name="Alice", updating_user=UserInfo.dummy_bob(), event_info=EventInfo.dummy(), updated_items=[] 

837 ) 

838 return [ 

839 replace(prototype, updated_items=[]), 

840 replace(prototype, updated_items=notification_data_pb2.EventUpdateItem.values()), 

841 ] 

842 

843 

844@dataclass(kw_only=True, slots=True) 

845class EventOrganizerInvitedEmail(EmailBase): 

846 """Sent when a user is invited to co-organize an event.""" 

847 

848 inviting_user: UserInfo 

849 event_info: EventInfo 

850 

851 @property 

852 def string_key_base(self) -> str: 

853 return "events.organizer_invited" 

854 

855 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

856 return self._localize( 

857 loc_context, 

858 ".subject", 

859 {"user": self.inviting_user.name, "title": self.event_info.title}, 

860 ) 

861 

862 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

863 builder = self._body_builder(loc_context) 

864 builder.para(".body", {"user": self.inviting_user.name, "title": self.event_info.title}) 

865 builder.block(self.event_info.get_details_block(loc_context)) 

866 builder.user(self.inviting_user, comment_key=".user_card_text") 

867 builder.block(self.event_info.get_description_block()) 

868 builder.block(self.event_info.get_view_action_block(loc_context)) 

869 builder.para(_do_not_reply_request_string_key) 

870 return builder.build() 

871 

872 @classmethod 

873 def from_notification(cls, data: notification_data_pb2.EventInviteOrganizer, *, user_name: str) -> Self: 

874 return cls( 

875 user_name=user_name, 

876 inviting_user=UserInfo.from_protobuf(data.inviting_user), 

877 event_info=EventInfo.from_proto(data.event), 

878 ) 

879 

880 @classmethod 

881 def test_instances(cls) -> list[Self]: 

882 return [cls(user_name="Alice", inviting_user=UserInfo.dummy_bob(), event_info=EventInfo.dummy())] 

883 

884 

885@dataclass(kw_only=True, slots=True) 

886class EventCommentEmail(EmailBase): 

887 """Sent to subscribers when someone comments on an event.""" 

888 

889 author: UserInfo 

890 event_info: EventInfo 

891 comment_markdown: str 

892 

893 @property 

894 def string_key_base(self) -> str: 

895 return "events.comment" 

896 

897 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

898 return self._localize(loc_context, ".subject", {"author": self.author.name, "title": self.event_info.title}) 

899 

900 def get_preview_line(self, loc_context: LocalizationContext) -> str | None: 

901 return markdown_to_plaintext(self.comment_markdown) 

902 

903 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

904 builder = self._body_builder(loc_context) 

905 builder.para(".body", {"author": self.author.name, "title": self.event_info.title}) 

906 builder.user(self.author) 

907 builder.quote(self.comment_markdown, markdown=True) 

908 builder.para(".event_details") 

909 builder.block(self.event_info.get_details_block(loc_context)) 

910 builder.block(self.event_info.get_view_action_block(loc_context)) 

911 return builder.build() 

912 

913 @classmethod 

914 def from_notification(cls, data: notification_data_pb2.EventComment, *, user_name: str) -> Self: 

915 return cls( 

916 user_name=user_name, 

917 author=UserInfo.from_protobuf(data.author), 

918 event_info=EventInfo.from_proto(data.event), 

919 comment_markdown=data.reply.content, 

920 ) 

921 

922 @classmethod 

923 def test_instances(cls) -> list[Self]: 

924 return [ 

925 cls( 

926 user_name="Alice", 

927 author=UserInfo.dummy_bob(), 

928 event_info=EventInfo.dummy(), 

929 comment_markdown="Looking forward to it, see you all there!", 

930 ) 

931 ] 

932 

933 

934@dataclass(kw_only=True, slots=True) 

935class EventReminderEmail(EmailBase): 

936 """Sent to subscribers as a reminder that an event starts soon.""" 

937 

938 event_info: EventInfo 

939 

940 @property 

941 def string_key_base(self) -> str: 

942 return "events.reminder" 

943 

944 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

945 return self._localize(loc_context, ".subject", {"title": self.event_info.title}) 

946 

947 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

948 builder = self._body_builder(loc_context) 

949 builder.para(".body") 

950 builder.block(self.event_info.get_details_block(loc_context)) 

951 builder.block(self.event_info.get_description_block()) 

952 builder.block(self.event_info.get_view_action_block(loc_context)) 

953 return builder.build() 

954 

955 @classmethod 

956 def from_notification(cls, data: notification_data_pb2.EventReminder, *, user_name: str) -> Self: 

957 return cls( 

958 user_name=user_name, 

959 event_info=EventInfo.from_proto(data.event), 

960 ) 

961 

962 @classmethod 

963 def test_instances(cls) -> list[Self]: 

964 return [cls(user_name="Alice", event_info=EventInfo.dummy())] 

965 

966 

967@dataclass(kw_only=True, slots=True) 

968class EventCancelledEmail(EmailBase): 

969 """Sent to subscribers when an event is cancelled.""" 

970 

971 cancelling_user: UserInfo 

972 event_info: EventInfo 

973 

974 @property 

975 def string_key_base(self) -> str: 

976 return "events.cancel" 

977 

978 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

979 return self._localize( 

980 loc_context, 

981 ".subject", 

982 {"user": self.cancelling_user.name, "title": self.event_info.title}, 

983 ) 

984 

985 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

986 builder = self._body_builder(loc_context) 

987 builder.para(".body") 

988 builder.block(self.event_info.get_details_block(loc_context)) 

989 builder.user(self.cancelling_user, ".user_card_text") 

990 builder.quote(self.event_info.description_markdown, markdown=True) 

991 builder.block(self.event_info.get_view_action_block(loc_context)) 

992 return builder.build() 

993 

994 @classmethod 

995 def from_notification(cls, data: notification_data_pb2.EventCancel, *, user_name: str) -> Self: 

996 return cls( 

997 user_name=user_name, 

998 cancelling_user=UserInfo.from_protobuf(data.cancelling_user), 

999 event_info=EventInfo.from_proto(data.event), 

1000 ) 

1001 

1002 @classmethod 

1003 def test_instances(cls) -> list[Self]: 

1004 return [cls(user_name="Alice", cancelling_user=UserInfo.dummy_bob(), event_info=EventInfo.dummy())] 

1005 

1006 

1007@dataclass(kw_only=True, slots=True) 

1008class EventDeletedEmail(EmailBase): 

1009 """Sent to subscribers when a moderator deletes an event.""" 

1010 

1011 event_info: EventInfo 

1012 

1013 @property 

1014 def string_key_base(self) -> str: 

1015 return "events.deleted" 

1016 

1017 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1018 return self._localize(loc_context, ".subject", {"title": self.event_info.title}) 

1019 

1020 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1021 builder = self._body_builder(loc_context) 

1022 builder.para(".body") 

1023 builder.block(self.event_info.get_details_block(loc_context)) 

1024 return builder.build() 

1025 

1026 @classmethod 

1027 def from_notification(cls, data: notification_data_pb2.EventDelete, *, user_name: str) -> Self: 

1028 return cls( 

1029 user_name=user_name, 

1030 event_info=EventInfo.from_proto(data.event), 

1031 ) 

1032 

1033 @classmethod 

1034 def test_instances(cls) -> list[Self]: 

1035 return [ 

1036 cls( 

1037 user_name="Alice", 

1038 event_info=EventInfo.dummy(), 

1039 ) 

1040 ] 

1041 

1042 

1043@dataclass(kw_only=True, slots=True) 

1044class FriendReferenceReceivedEmail(EmailBase): 

1045 """Sent to a user when they receive a friend reference.""" 

1046 

1047 from_user: UserInfo 

1048 text: str 

1049 

1050 @property 

1051 def string_key_base(self) -> str: 

1052 return "references.received.friend" 

1053 

1054 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1055 return self._localize(loc_context, ".subject", {"name": self.from_user.name}) 

1056 

1057 def get_preview_line(self, loc_context: LocalizationContext) -> str | None: 

1058 return self.text 

1059 

1060 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1061 builder = self._body_builder(loc_context) 

1062 builder.para(".body", {"name": self.from_user.name}) 

1063 builder.user(self.from_user) 

1064 builder.quote(self.text, markdown=False) 

1065 builder.action(urls.profile_references_link(), "references.received.view_action") 

1066 return builder.build() 

1067 

1068 @classmethod 

1069 def from_notification(cls, data: notification_data_pb2.ReferenceReceiveFriend, *, user_name: str) -> Self: 

1070 return cls(user_name=user_name, from_user=UserInfo.from_protobuf(data.from_user), text=data.text) 

1071 

1072 @classmethod 

1073 def test_instances(cls) -> list[Self]: 

1074 return [ 

1075 cls( 

1076 user_name="Alice", 

1077 from_user=UserInfo.dummy_bob(), 

1078 text="Alice is a wonderful person and a great travel companion!", 

1079 ) 

1080 ] 

1081 

1082 

1083@dataclass(kw_only=True, slots=True) 

1084class FriendRequestReceivedEmail(EmailBase): 

1085 """Sent to a user when they receive a friend request.""" 

1086 

1087 befriender: UserInfo 

1088 

1089 @property 

1090 def string_key_base(self) -> str: 

1091 return "friend_requests.received" 

1092 

1093 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1094 return self._localize(loc_context, ".subject", {"name": self.befriender.name}) 

1095 

1096 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1097 builder = self._body_builder(loc_context) 

1098 builder.para(".body", {"name": self.befriender.name}) 

1099 builder.user(self.befriender) 

1100 builder.action(urls.friend_requests_link(), ".view_action") 

1101 builder.para(".closing") 

1102 builder.para(_do_not_reply_request_string_key) 

1103 return builder.build() 

1104 

1105 @classmethod 

1106 def from_notification(cls, data: notification_data_pb2.FriendRequestCreate, *, user_name: str) -> Self: 

1107 return cls(user_name=user_name, befriender=UserInfo.from_protobuf(data.other_user)) 

1108 

1109 @classmethod 

1110 def test_instances(cls) -> list[Self]: 

1111 return [ 

1112 cls( 

1113 user_name="Alice", 

1114 befriender=UserInfo.dummy_bob(), 

1115 ) 

1116 ] 

1117 

1118 

1119@dataclass(kw_only=True, slots=True) 

1120class FriendRequestAcceptedEmail(EmailBase): 

1121 """Sent to a user when their friend request is accepted.""" 

1122 

1123 new_friend: UserInfo 

1124 

1125 @property 

1126 def string_key_base(self) -> str: 

1127 return "friend_requests.accepted" 

1128 

1129 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1130 return self._localize(loc_context, ".subject", {"name": self.new_friend.name}) 

1131 

1132 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1133 builder = self._body_builder(loc_context) 

1134 builder.para(".body", {"name": self.new_friend.name}) 

1135 builder.user(self.new_friend) 

1136 builder.action(self.new_friend.profile_url, ".view_action") 

1137 builder.para(".closing") 

1138 return builder.build() 

1139 

1140 @classmethod 

1141 def from_notification(cls, data: notification_data_pb2.FriendRequestAccept, *, user_name: str) -> Self: 

1142 return cls(user_name=user_name, new_friend=UserInfo.from_protobuf(data.other_user)) 

1143 

1144 @classmethod 

1145 def test_instances(cls) -> list[Self]: 

1146 return [ 

1147 cls( 

1148 user_name="Alice", 

1149 new_friend=UserInfo.dummy_bob(), 

1150 ) 

1151 ] 

1152 

1153 

1154@dataclass(kw_only=True, slots=True) 

1155class GenderChangedEmail(EmailBase): 

1156 """Sent to a user to notify them that their gender was changed.""" 

1157 

1158 new_gender: str 

1159 

1160 @property 

1161 def string_key_base(self) -> str: 

1162 return "gender_changed" 

1163 

1164 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1165 builder = self._body_builder(loc_context, security_warning=True) 

1166 builder.para(".body", {"gender": self.new_gender}) 

1167 return builder.build() 

1168 

1169 @classmethod 

1170 def from_notification(cls, data: notification_data_pb2.GenderChange, *, user_name: str) -> Self: 

1171 return cls(user_name=user_name, new_gender=data.gender) 

1172 

1173 @classmethod 

1174 def test_instances(cls) -> list[Self]: 

1175 return [ 

1176 cls( 

1177 user_name="Alice", 

1178 new_gender="Male", 

1179 ) 

1180 ] 

1181 

1182 

1183@dataclass(kw_only=True, slots=True) 

1184class HostRequestCreatedEmail(EmailBase): 

1185 """Sent to a host when a surfer sends them a new host request.""" 

1186 

1187 surfer: UserInfo 

1188 from_date: date 

1189 to_date: date 

1190 text: str 

1191 quick_decline_link: str 

1192 view_link: str 

1193 

1194 @property 

1195 def string_key_base(self) -> str: 

1196 return "host_requests.created" 

1197 

1198 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1199 return self._localize(loc_context, ".subject", {"surfer_name": self.surfer.name}) 

1200 

1201 def get_preview_line(self, loc_context: LocalizationContext) -> str | None: 

1202 return self.text 

1203 

1204 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1205 builder = self._body_builder(loc_context) 

1206 builder.para(".body", {"surfer_name": self.surfer.name}) 

1207 builder.user( 

1208 self.surfer, 

1209 "host_requests.generic.date_range", 

1210 { 

1211 "from_date": _localize_host_request_date(self.from_date, loc_context), 

1212 "to_date": _localize_host_request_date(self.to_date, loc_context), 

1213 }, 

1214 ) 

1215 builder.quote(self.text, markdown=False) 

1216 builder.action(self.view_link, "host_requests.generic.view_action") 

1217 builder.action(self.quick_decline_link, ".quick_decline_action") 

1218 builder.para(".respond_encouragement") 

1219 builder.para(_do_not_reply_request_string_key) 

1220 return builder.build() 

1221 

1222 @classmethod 

1223 def from_notification(cls, data: notification_data_pb2.HostRequestCreate, *, user_name: str) -> Self: 

1224 return cls( 

1225 user_name, 

1226 surfer=UserInfo.from_protobuf(data.surfer), 

1227 from_date=date.fromisoformat(data.host_request.from_date), 

1228 to_date=date.fromisoformat(data.host_request.to_date), 

1229 text=data.text, 

1230 quick_decline_link=generate_quick_decline_link(data.host_request), 

1231 view_link=urls.host_request(host_request_id=data.host_request.host_request_id), 

1232 ) 

1233 

1234 @classmethod 

1235 def test_instances(cls) -> list[Self]: 

1236 return [ 

1237 cls( 

1238 user_name="Alice", 

1239 surfer=UserInfo.dummy_bob(), 

1240 from_date=date(2025, 6, 1), 

1241 to_date=date(2025, 6, 7), 

1242 text="Hey, I'd love to stay for a few nights!", 

1243 quick_decline_link="https://couchers.org/requests/123/decline?token=xxx", 

1244 view_link="https://couchers.org/requests/123", 

1245 ) 

1246 ] 

1247 

1248 

1249@dataclass(kw_only=True, slots=True) 

1250class HostRequestReminderEmail(EmailBase): 

1251 """Sent to a host as a reminder to respond to a pending host request.""" 

1252 

1253 surfer: UserInfo 

1254 from_date: date 

1255 to_date: date 

1256 view_link: str 

1257 

1258 @property 

1259 def string_key_base(self) -> str: 

1260 return "host_requests.reminder" 

1261 

1262 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1263 return self._localize(loc_context, ".subject", {"surfer_name": self.surfer.name}) 

1264 

1265 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1266 builder = self._body_builder(loc_context) 

1267 builder.para(".body") 

1268 builder.user( 

1269 self.surfer, 

1270 "host_requests.generic.date_range", 

1271 { 

1272 "from_date": _localize_host_request_date(self.from_date, loc_context), 

1273 "to_date": _localize_host_request_date(self.to_date, loc_context), 

1274 }, 

1275 ) 

1276 builder.action(self.view_link, "host_requests.generic.view_action") 

1277 builder.para(_do_not_reply_request_string_key) 

1278 return builder.build() 

1279 

1280 @classmethod 

1281 def from_notification(cls, data: notification_data_pb2.HostRequestReminder, *, user_name: str) -> Self: 

1282 return cls( 

1283 user_name, 

1284 surfer=UserInfo.from_protobuf(data.surfer), 

1285 from_date=date.fromisoformat(data.host_request.from_date), 

1286 to_date=date.fromisoformat(data.host_request.to_date), 

1287 view_link=urls.host_request(host_request_id=data.host_request.host_request_id), 

1288 ) 

1289 

1290 @classmethod 

1291 def test_instances(cls) -> list[Self]: 

1292 return [ 

1293 cls( 

1294 user_name="Alice", 

1295 surfer=UserInfo.dummy_bob(), 

1296 from_date=date(2025, 6, 1), 

1297 to_date=date(2025, 6, 7), 

1298 view_link="https://couchers.org/requests/123", 

1299 ) 

1300 ] 

1301 

1302 

1303@dataclass(kw_only=True, slots=True) 

1304class HostRequestMessageEmail(EmailBase): 

1305 """Sent when a user sends a message in an existing host request.""" 

1306 

1307 other_user: UserInfo 

1308 from_date: date 

1309 to_date: date 

1310 text: str 

1311 from_host: bool 

1312 view_link: str 

1313 

1314 @property 

1315 def string_key_base(self) -> str: 

1316 variant = "from_host" if self.from_host else "from_surfer" 

1317 return f"host_requests.message.{variant}" 

1318 

1319 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1320 return self._localize(loc_context, ".subject", {"other_name": self.other_user.name}) 

1321 

1322 def get_preview_line(self, loc_context: LocalizationContext) -> str | None: 

1323 return self.text 

1324 

1325 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1326 builder = self._body_builder(loc_context) 

1327 builder.para(".body", {"other_name": self.other_user.name}) 

1328 builder.user( 

1329 self.other_user, 

1330 "host_requests.generic.date_range", 

1331 { 

1332 "from_date": _localize_host_request_date(self.from_date, loc_context), 

1333 "to_date": _localize_host_request_date(self.to_date, loc_context), 

1334 }, 

1335 ) 

1336 builder.quote(self.text, markdown=False) 

1337 builder.action(self.view_link, "host_requests.generic.view_action") 

1338 builder.para(_do_not_reply_request_string_key) 

1339 return builder.build() 

1340 

1341 @classmethod 

1342 def from_notification(cls, data: notification_data_pb2.HostRequestMessage, *, user_name: str) -> Self: 

1343 return cls( 

1344 user_name, 

1345 other_user=UserInfo.from_protobuf(data.user), 

1346 from_date=date.fromisoformat(data.host_request.from_date), 

1347 to_date=date.fromisoformat(data.host_request.to_date), 

1348 text=data.text, 

1349 from_host=not data.am_host, 

1350 view_link=urls.host_request(host_request_id=data.host_request.host_request_id), 

1351 ) 

1352 

1353 @classmethod 

1354 def test_instances(cls) -> list[Self]: 

1355 prototype = cls( 

1356 user_name="Alice", 

1357 other_user=UserInfo.dummy_bob(), 

1358 from_date=date(2025, 6, 1), 

1359 to_date=date(2025, 6, 7), 

1360 text="Looking forward to it, see you soon!", 

1361 from_host=True, 

1362 view_link="https://couchers.org/requests/123", 

1363 ) 

1364 return [replace(prototype, from_host=True), replace(prototype, from_host=False)] 

1365 

1366 

1367@dataclass(kw_only=True, slots=True) 

1368class HostRequestMissedMessagesEmail(EmailBase): 

1369 """Sent as a digest when a user has missed messages in a host request.""" 

1370 

1371 other_user: UserInfo 

1372 from_date: date 

1373 to_date: date 

1374 from_host: bool 

1375 view_link: str 

1376 

1377 @property 

1378 def string_key_base(self) -> str: 

1379 variant = "from_host" if self.from_host else "from_surfer" 

1380 return f"host_requests.missed_messages.{variant}" 

1381 

1382 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1383 return self._localize(loc_context, ".subject", {"other_name": self.other_user.name}) 

1384 

1385 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1386 builder = self._body_builder(loc_context) 

1387 builder.para(".body", {"other_name": self.other_user.name}) 

1388 builder.user( 

1389 self.other_user, 

1390 "host_requests.generic.date_range", 

1391 { 

1392 "from_date": _localize_host_request_date(self.from_date, loc_context), 

1393 "to_date": _localize_host_request_date(self.to_date, loc_context), 

1394 }, 

1395 ) 

1396 builder.action(self.view_link, "host_requests.generic.view_action") 

1397 builder.para(_do_not_reply_request_string_key) 

1398 return builder.build() 

1399 

1400 @classmethod 

1401 def from_notification(cls, data: notification_data_pb2.HostRequestMissedMessages, *, user_name: str) -> Self: 

1402 return cls( 

1403 user_name, 

1404 other_user=UserInfo.from_protobuf(data.user), 

1405 from_date=date.fromisoformat(data.host_request.from_date), 

1406 to_date=date.fromisoformat(data.host_request.to_date), 

1407 from_host=not data.am_host, 

1408 view_link=urls.host_request(host_request_id=data.host_request.host_request_id), 

1409 ) 

1410 

1411 @classmethod 

1412 def test_instances(cls) -> list[Self]: 

1413 prototype = cls( 

1414 user_name="Alice", 

1415 other_user=UserInfo.dummy_bob(), 

1416 from_date=date(2025, 6, 1), 

1417 to_date=date(2025, 6, 7), 

1418 from_host=True, 

1419 view_link="https://couchers.org/requests/123", 

1420 ) 

1421 return [replace(prototype, from_host=True), replace(prototype, from_host=False)] 

1422 

1423 

1424@dataclass(kw_only=True, slots=True) 

1425class HostRequestStatusChangedEmail(EmailBase): 

1426 """Sent when a host request is accepted, declined, confirmed, or cancelled.""" 

1427 

1428 other_user: UserInfo 

1429 from_date: date 

1430 to_date: date 

1431 new_status: conversations_pb2.HostRequestStatus.ValueType 

1432 view_link: str 

1433 

1434 @property 

1435 def string_key_base(self) -> str: 

1436 base_key = "host_requests.status_changed" 

1437 match self.new_status: 

1438 case conversations_pb2.HOST_REQUEST_STATUS_ACCEPTED: 

1439 return f"{base_key}.accepted_by_host" 

1440 case conversations_pb2.HOST_REQUEST_STATUS_REJECTED: 

1441 return f"{base_key}.declined_by_host" 

1442 case conversations_pb2.HOST_REQUEST_STATUS_CONFIRMED: 

1443 return f"{base_key}.confirmed_by_surfer" 

1444 case conversations_pb2.HOST_REQUEST_STATUS_CANCELLED: 1444 ↛ 1446line 1444 didn't jump to line 1446 because the pattern on line 1444 always matched

1445 return f"{base_key}.cancelled_by_surfer" 

1446 case _: 

1447 raise ValueError(f"Unexpected host request status: {self.new_status}") 

1448 

1449 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1450 return self._localize(loc_context, ".subject", {"other_name": self.other_user.name}) 

1451 

1452 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1453 builder = self._body_builder(loc_context) 

1454 builder.para(".body", {"other_name": self.other_user.name}) 

1455 builder.user( 

1456 self.other_user, 

1457 "host_requests.generic.date_range", 

1458 { 

1459 "from_date": _localize_host_request_date(self.from_date, loc_context), 

1460 "to_date": _localize_host_request_date(self.to_date, loc_context), 

1461 }, 

1462 ) 

1463 builder.action(self.view_link, "host_requests.generic.view_action") 

1464 builder.para(_do_not_reply_request_string_key) 

1465 return builder.build() 

1466 

1467 @classmethod 

1468 def from_notification( 

1469 cls, 

1470 data: notification_data_pb2.HostRequestAccept 

1471 | notification_data_pb2.HostRequestReject 

1472 | notification_data_pb2.HostRequestConfirm 

1473 | notification_data_pb2.HostRequestCancel, 

1474 *, 

1475 user_name: str, 

1476 ) -> Self: 

1477 other_user: UserInfo 

1478 new_status: conversations_pb2.HostRequestStatus.ValueType 

1479 match data: 

1480 case notification_data_pb2.HostRequestAccept(): 

1481 other_user = UserInfo.from_protobuf(data.host) 

1482 new_status = conversations_pb2.HostRequestStatus.HOST_REQUEST_STATUS_ACCEPTED 

1483 case notification_data_pb2.HostRequestReject(): 1483 ↛ 1484line 1483 didn't jump to line 1484 because the pattern on line 1483 never matched

1484 other_user = UserInfo.from_protobuf(data.host) 

1485 new_status = conversations_pb2.HostRequestStatus.HOST_REQUEST_STATUS_REJECTED 

1486 case notification_data_pb2.HostRequestConfirm(): 

1487 other_user = UserInfo.from_protobuf(data.surfer) 

1488 new_status = conversations_pb2.HostRequestStatus.HOST_REQUEST_STATUS_CONFIRMED 

1489 case notification_data_pb2.HostRequestCancel(): 1489 ↛ 1492line 1489 didn't jump to line 1492 because the pattern on line 1489 always matched

1490 other_user = UserInfo.from_protobuf(data.surfer) 

1491 new_status = conversations_pb2.HostRequestStatus.HOST_REQUEST_STATUS_CANCELLED 

1492 case _: 

1493 # Enable mypy's exhaustiveness checking 

1494 assert_never("Unexpected host request status changed notification data type.") 

1495 

1496 return cls( 

1497 user_name, 

1498 other_user=other_user, 

1499 from_date=date.fromisoformat(data.host_request.from_date), 

1500 to_date=date.fromisoformat(data.host_request.to_date), 

1501 new_status=new_status, 

1502 view_link=urls.host_request(host_request_id=data.host_request.host_request_id), 

1503 ) 

1504 

1505 @classmethod 

1506 def test_instances(cls) -> list[Self]: 

1507 prototype = cls( 

1508 user_name="Alice", 

1509 other_user=UserInfo.dummy_bob(), 

1510 from_date=date(2025, 6, 1), 

1511 to_date=date(2025, 6, 7), 

1512 new_status=conversations_pb2.HOST_REQUEST_STATUS_ACCEPTED, 

1513 view_link="https://couchers.org/requests/123", 

1514 ) 

1515 return [ 

1516 replace(prototype, new_status=conversations_pb2.HOST_REQUEST_STATUS_ACCEPTED), 

1517 replace(prototype, new_status=conversations_pb2.HOST_REQUEST_STATUS_REJECTED), 

1518 replace(prototype, new_status=conversations_pb2.HOST_REQUEST_STATUS_CONFIRMED), 

1519 replace(prototype, new_status=conversations_pb2.HOST_REQUEST_STATUS_CANCELLED), 

1520 ] 

1521 

1522 

1523@dataclass(kw_only=True, slots=True) 

1524class HostReferenceReceivedEmail(EmailBase): 

1525 """Sent to a user when they receive a reference from a past host or surfer.""" 

1526 

1527 from_user: UserInfo 

1528 text: str | None # None if hidden because receiver hasn't written their reference yet. 

1529 surfed: bool # True if I was the surfer, False if I was the host 

1530 leave_reference_url: str 

1531 

1532 @property 

1533 def string_key_base(self) -> str: 

1534 return "references.received" 

1535 

1536 @property 

1537 def string_role_subkey(self) -> str: 

1538 return "surfed" if self.surfed else "hosted" 

1539 

1540 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1541 return self._localize(loc_context, ".subject", {"name": self.from_user.name}) 

1542 

1543 def get_preview_line(self, loc_context: LocalizationContext) -> str | None: 

1544 return self.text 

1545 

1546 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1547 builder = self._body_builder(loc_context) 

1548 builder.para(f".{self.string_role_subkey}.body", {"name": self.from_user.name}) 

1549 builder.user(self.from_user) 

1550 if self.text: 

1551 builder.para(".before_quote") 

1552 builder.quote(self.text, markdown=False) 

1553 builder.action(urls.profile_references_link(), ".view_action") 

1554 else: 

1555 builder.para(".reciprocate_encouragement", {"name": self.from_user.name}) 

1556 builder.action(self.leave_reference_url, "references.write_action", {"name": self.from_user.name}) 

1557 return builder.build() 

1558 

1559 @classmethod 

1560 def from_notification( 

1561 cls, data: notification_data_pb2.ReferenceReceiveHostRequest, *, user_name: str, surfed: bool 

1562 ) -> Self: 

1563 return cls( 

1564 user_name=user_name, 

1565 from_user=UserInfo.from_protobuf(data.from_user), 

1566 text=data.text or None, 

1567 surfed=surfed, 

1568 leave_reference_url=urls.leave_reference_link( 

1569 reference_type="surfed" if surfed else "hosted", 

1570 to_user_id=str(data.from_user.user_id), 

1571 host_request_id=str(data.host_request_id), 

1572 ), 

1573 ) 

1574 

1575 @classmethod 

1576 def test_instances(cls) -> list[Self]: 

1577 prototype = cls( 

1578 user_name="Alice", 

1579 from_user=UserInfo.dummy_bob(), 

1580 text="Alice was a fantastic guest!", 

1581 surfed=True, 

1582 leave_reference_url="https://couchers.org/leave-reference/123", 

1583 ) 

1584 return [ 

1585 replace(prototype, surfed=True, text="Alice was a fantastic guest!"), 

1586 replace(prototype, surfed=True, text=None), 

1587 replace(prototype, surfed=False, text="Bob was a wonderful host!"), 

1588 replace(prototype, surfed=False, text=None), 

1589 ] 

1590 

1591 

1592@dataclass(kw_only=True, slots=True) 

1593class HostReferenceReminderEmail(EmailBase): 

1594 """Sent as a reminder to write a reference after a stay.""" 

1595 

1596 other_user: UserInfo 

1597 days_left: int 

1598 surfed: bool # True if I was the surfer, False if I was the host 

1599 leave_reference_url: str 

1600 

1601 @property 

1602 def string_key_base(self) -> str: 

1603 return "references.reminder" 

1604 

1605 @property 

1606 def string_role_subkey(self) -> str: 

1607 return "surfed" if self.surfed else "hosted" 

1608 

1609 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1610 return self._localize( 

1611 loc_context, 

1612 ".subject_days", 

1613 {"name": self.other_user.name, "count": self.days_left}, 

1614 ) 

1615 

1616 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1617 builder = self._body_builder(loc_context) 

1618 builder.para(f".{self.string_role_subkey}.body_days", {"name": self.other_user.name, "count": self.days_left}) 

1619 builder.para(".no_meeting_note", {"name": self.other_user.name}) 

1620 builder.user(self.other_user) 

1621 builder.action( 

1622 self.leave_reference_url, 

1623 "references.write_action", 

1624 {"name": self.other_user.name}, 

1625 ) 

1626 builder.para(".via_messaging_note") 

1627 builder.para(".importance_note") 

1628 builder.para(".visibility_note") 

1629 return builder.build() 

1630 

1631 @classmethod 

1632 def from_notification(cls, data: notification_data_pb2.ReferenceReminder, *, user_name: str, surfed: bool) -> Self: 

1633 return cls( 

1634 user_name=user_name, 

1635 other_user=UserInfo.from_protobuf(data.other_user), 

1636 days_left=data.days_left, 

1637 surfed=surfed, 

1638 leave_reference_url=urls.leave_reference_link( 

1639 reference_type="surfed" if surfed else "hosted", 

1640 to_user_id=str(data.other_user.user_id), 

1641 host_request_id=str(data.host_request_id), 

1642 ), 

1643 ) 

1644 

1645 @classmethod 

1646 def test_instances(cls) -> list[Self]: 

1647 prototype = cls( 

1648 user_name="Alice", 

1649 other_user=UserInfo.dummy_bob(), 

1650 days_left=7, 

1651 surfed=True, 

1652 leave_reference_url="https://couchers.org/leave-reference/123", 

1653 ) 

1654 return [replace(prototype, surfed=True), replace(prototype, surfed=False)] 

1655 

1656 

1657@dataclass(kw_only=True, slots=True) 

1658class ModeratorNoteEmail(EmailBase): 

1659 """Sent to a user to notify them they have received a moderator note.""" 

1660 

1661 @property 

1662 def string_key_base(self) -> str: 

1663 return "moderator_note" 

1664 

1665 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1666 builder = self._body_builder(loc_context) 

1667 builder.para(".body") 

1668 return builder.build() 

1669 

1670 @classmethod 

1671 def test_instances(cls) -> list[Self]: 

1672 return [cls(user_name="Alice")] 

1673 

1674 

1675@dataclass(kw_only=True, slots=True) 

1676class NewBlogPostEmail(EmailBase): 

1677 """Sent to notify users of a new blog post.""" 

1678 

1679 title: str 

1680 blurb: str 

1681 url: str 

1682 

1683 @property 

1684 def string_key_base(self) -> str: 

1685 return "new_blog_post" 

1686 

1687 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1688 return self._localize(loc_context, ".subject", {"title": self.title}) 

1689 

1690 def get_preview_line(self, loc_context: LocalizationContext) -> str | None: 

1691 return self.blurb 

1692 

1693 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1694 builder = self._body_builder(loc_context) 

1695 builder.para(".intro") 

1696 builder.para(".post_title", {"title": self.title}) 

1697 builder.quote(self.blurb, markdown=False) 

1698 builder.action(self.url, ".read_action") 

1699 return builder.build() 

1700 

1701 @classmethod 

1702 def from_notification(cls, data: notification_data_pb2.GeneralNewBlogPost, *, user_name: str) -> Self: 

1703 return cls(user_name=user_name, title=data.title, blurb=data.blurb, url=data.url) 

1704 

1705 @classmethod 

1706 def test_instances(cls) -> list[Self]: 

1707 return [ 

1708 cls( 

1709 user_name="Alice", 

1710 title="Exciting new features on Couchers.org", 

1711 blurb="We've launched some great new features including improved messaging and event discovery.", 

1712 url="https://couchers.org/blog/2025/01/01/new-features", 

1713 ) 

1714 ] 

1715 

1716 

1717@dataclass(kw_only=True, slots=True) 

1718class OnboardingReminderEmail(EmailBase): 

1719 """Onboarding email sent to new users; initial=True for the first email, False for the second.""" 

1720 

1721 initial: bool 

1722 

1723 @property 

1724 def string_key_base(self) -> str: 

1725 return f"onboarding_reminder.{'initial' if self.initial else 'follow_up'}" 

1726 

1727 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1728 builder = self._body_builder(loc_context, standard_closing=False) 

1729 edit_profile_url = urls.edit_profile_link() 

1730 if self.initial: 

1731 builder.para(".welcome") 

1732 builder.para(".early_user_role") 

1733 builder.para(".fill_in_profile") 

1734 builder.para(".edit_profile_prompt") 

1735 builder.action(edit_profile_url, ".edit_profile_action") 

1736 builder.para(".share_with_friends") 

1737 builder.para(".link", {"url": urls.app_link()}) 

1738 builder.para(".platform_under_development") 

1739 builder.para(".thanks_for_joining") 

1740 builder.para(".signature") 

1741 else: 

1742 builder.para(".intro") 

1743 builder.para(".fill_in_profile") 

1744 builder.action(edit_profile_url, ".edit_profile_action") 

1745 builder.para(".no_empty_accounts") 

1746 builder.para(".profile_importance") 

1747 builder.para(".signature") 

1748 return builder.build() 

1749 

1750 @classmethod 

1751 def test_instances(cls) -> list[Self]: 

1752 prototype = cls(user_name="Alice", initial=True) 

1753 return [replace(prototype, initial=True), replace(prototype, initial=False)] 

1754 

1755 

1756@dataclass(kw_only=True, slots=True) 

1757class PasswordChangedEmail(EmailBase): 

1758 """Sent to a user to notify them that their login password was changed.""" 

1759 

1760 @property 

1761 def string_key_base(self) -> str: 

1762 return "password_changed" 

1763 

1764 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1765 builder = self._body_builder(loc_context, security_warning=True) 

1766 builder.para(".body") 

1767 return builder.build() 

1768 

1769 @classmethod 

1770 def test_instances(cls) -> list[Self]: 

1771 return [cls(user_name="Alice")] 

1772 

1773 

1774@dataclass(kw_only=True, slots=True) 

1775class PasswordResetCompletedEmail(EmailBase): 

1776 """Sent to a user to confirm their password was successfully reset.""" 

1777 

1778 @property 

1779 def string_key_base(self) -> str: 

1780 return "password_reset.completed" 

1781 

1782 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1783 builder = self._body_builder(loc_context, security_warning=True) 

1784 builder.para(".body") 

1785 return builder.build() 

1786 

1787 @classmethod 

1788 def test_instances(cls) -> list[Self]: 

1789 return [cls(user_name="Alice")] 

1790 

1791 

1792@dataclass(kw_only=True, slots=True) 

1793class PasswordResetStartedEmail(EmailBase): 

1794 """Sent to a user with a link to complete their password reset.""" 

1795 

1796 password_reset_link: str 

1797 

1798 @property 

1799 def string_key_base(self) -> str: 

1800 return "password_reset.started" 

1801 

1802 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1803 builder = self._body_builder(loc_context, security_warning=True) 

1804 builder.para(".request_description") 

1805 builder.para(".confirmation_instructions") 

1806 builder.action(self.password_reset_link, ".reset_action") 

1807 return builder.build() 

1808 

1809 @classmethod 

1810 def from_notification(cls, data: notification_data_pb2.PasswordResetStart, *, user_name: str) -> Self: 

1811 return cls( 

1812 user_name=user_name, 

1813 password_reset_link=urls.password_reset_link(password_reset_token=data.password_reset_token), 

1814 ) 

1815 

1816 @classmethod 

1817 def test_instances(cls) -> list[Self]: 

1818 return [cls(user_name="Alice", password_reset_link="https://couchers.org/reset-password")] 

1819 

1820 

1821@dataclass(kw_only=True, slots=True) 

1822class PhoneNumberChangeEmail(EmailBase): 

1823 """Sent to a user to notify them that their phone number verification status was changed.""" 

1824 

1825 new_phone_number: str 

1826 completed: bool # False = started, True = completed 

1827 

1828 @property 

1829 def string_key_base(self) -> str: 

1830 return "phone_number_verification.verified" if self.completed else "phone_number_verification.started" 

1831 

1832 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1833 builder = self._body_builder(loc_context, security_warning=True) 

1834 builder.para(".body", {"phone_number": format_phone_number(self.new_phone_number)}) 

1835 return builder.build() 

1836 

1837 @classmethod 

1838 def from_change_notification(cls, data: notification_data_pb2.PhoneNumberChange, *, user_name: str) -> Self: 

1839 return cls(user_name=user_name, new_phone_number=data.phone, completed=False) 

1840 

1841 @classmethod 

1842 def from_verify_notification(cls, data: notification_data_pb2.PhoneNumberVerify, *, user_name: str) -> Self: 

1843 return cls(user_name=user_name, new_phone_number=data.phone, completed=True) 

1844 

1845 @classmethod 

1846 def test_instances(cls) -> list[Self]: 

1847 prototype = cls( 

1848 user_name="Alice", 

1849 new_phone_number="+12223334444", 

1850 completed=False, 

1851 ) 

1852 return [replace(prototype, completed=False), replace(prototype, completed=True)] 

1853 

1854 

1855@dataclass(kw_only=True, slots=True) 

1856class PostalVerificationFailedEmail(EmailBase): 

1857 """Sent to a user when their postal verification attempt has failed.""" 

1858 

1859 reason: notification_data_pb2.PostalVerificationFailReason.ValueType 

1860 

1861 @property 

1862 def string_key_base(self) -> str: 

1863 return "postal_verification.failed" 

1864 

1865 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1866 builder = self._body_builder(loc_context, security_warning=True) 

1867 match self.reason: 

1868 case notification_data_pb2.POSTAL_VERIFICATION_FAIL_REASON_CODE_EXPIRED: 

1869 reason_string_key = ".reason_code_expired" 

1870 case notification_data_pb2.POSTAL_VERIFICATION_FAIL_REASON_TOO_MANY_ATTEMPTS: 

1871 reason_string_key = ".reason_too_many_attempts" 

1872 case _: 

1873 reason_string_key = ".reason_unknown" 

1874 builder.para(reason_string_key) 

1875 return builder.build() 

1876 

1877 @classmethod 

1878 def from_notification(cls, data: notification_data_pb2.PostalVerificationFailed, *, user_name: str) -> Self: 

1879 return cls(user_name=user_name, reason=data.reason) 

1880 

1881 @classmethod 

1882 def test_instances(cls) -> list[Self]: 

1883 prototype = cls( 

1884 user_name="Alice", 

1885 reason=notification_data_pb2.POSTAL_VERIFICATION_FAIL_REASON_CODE_EXPIRED, 

1886 ) 

1887 return [ 

1888 replace(prototype, reason=notification_data_pb2.POSTAL_VERIFICATION_FAIL_REASON_CODE_EXPIRED), 

1889 replace(prototype, reason=notification_data_pb2.POSTAL_VERIFICATION_FAIL_REASON_TOO_MANY_ATTEMPTS), 

1890 replace(prototype, reason=notification_data_pb2.POSTAL_VERIFICATION_FAIL_REASON_UNKNOWN), 

1891 ] 

1892 

1893 

1894@dataclass(kw_only=True, slots=True) 

1895class PostalVerificationPostcardSentEmail(EmailBase): 

1896 """Sent to a user to notify them that their verification postcard has been sent.""" 

1897 

1898 city: str 

1899 country: str 

1900 

1901 @property 

1902 def string_key_base(self) -> str: 

1903 return "postal_verification.postcard_sent" 

1904 

1905 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1906 builder = self._body_builder(loc_context, security_warning=True) 

1907 builder.para(".body", {"city": self.city, "country": self.country}) 

1908 return builder.build() 

1909 

1910 @classmethod 

1911 def from_notification(cls, data: notification_data_pb2.PostalVerificationPostcardSent, *, user_name: str) -> Self: 

1912 return cls(user_name=user_name, city=data.city, country=data.country) 

1913 

1914 @classmethod 

1915 def test_instances(cls) -> list[Self]: 

1916 return [cls(user_name="Alice", city="New York", country="United States")] 

1917 

1918 

1919@dataclass(kw_only=True, slots=True) 

1920class PostalVerificationSucceededEmail(EmailBase): 

1921 """Sent to a user when their postal verification has succeeded.""" 

1922 

1923 @property 

1924 def string_key_base(self) -> str: 

1925 return "postal_verification.succeeded" 

1926 

1927 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1928 builder = self._body_builder(loc_context, security_warning=True) 

1929 builder.para(".body") 

1930 return builder.build() 

1931 

1932 @classmethod 

1933 def test_instances(cls) -> list[Self]: 

1934 return [cls(user_name="Alice")] 

1935 

1936 

1937@dataclass(kw_only=True, slots=True) 

1938class SignupVerifyEmail(EmailBase): 

1939 """Sent to a user to verify their email address.""" 

1940 

1941 verify_url: str 

1942 

1943 @property 

1944 def string_key_base(self) -> str: 

1945 return "signup.verify" 

1946 

1947 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1948 return self._localize(loc_context, "signup.subject") 

1949 

1950 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1951 builder = self._body_builder(loc_context) 

1952 builder.para(".thanks") 

1953 builder.para(".instructions") 

1954 builder.action(self.verify_url, ".confirm_action") 

1955 builder.para("signup.closing") 

1956 return builder.build() 

1957 

1958 @classmethod 

1959 def test_instances(cls) -> list[Self]: 

1960 return [cls(user_name="Alice", verify_url="https://example.com")] 

1961 

1962 

1963@dataclass(kw_only=True, slots=True) 

1964class SignupContinueEmail(EmailBase): 

1965 """Sent to a user to ask them to continue the signup process.""" 

1966 

1967 continue_url: str 

1968 

1969 @property 

1970 def string_key_base(self) -> str: 

1971 return "signup.continue" 

1972 

1973 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1974 return self._localize(loc_context, "signup.subject") 

1975 

1976 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1977 builder = self._body_builder(loc_context) 

1978 builder.para(".request") 

1979 builder.para(".instructions") 

1980 builder.action(self.continue_url, ".continue_action") 

1981 builder.para("signup.closing") 

1982 builder.para(".ignore_if_unexpected") 

1983 return builder.build() 

1984 

1985 @classmethod 

1986 def test_instances(cls) -> list[Self]: 

1987 return [cls(user_name="Alice", continue_url="https://example.com")] 

1988 

1989 

1990@dataclass(kw_only=True, slots=True) 

1991class StrongVerificationFailedEmail(EmailBase): 

1992 """Sent to a user when their strong verification attempt has failed.""" 

1993 

1994 reason: notification_data_pb2.SVFailReason.ValueType 

1995 

1996 @property 

1997 def string_key_base(self) -> str: 

1998 return "strong_verification.failed" 

1999 

2000 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

2001 builder = self._body_builder(loc_context, security_warning=True) 

2002 match self.reason: 

2003 case notification_data_pb2.SV_FAIL_REASON_WRONG_BIRTHDATE_OR_GENDER: 

2004 reason_string_key = ".reason_wrong_birthdate_or_gender" 

2005 case notification_data_pb2.SV_FAIL_REASON_NOT_A_PASSPORT: 

2006 reason_string_key = ".reason_not_a_passport" 

2007 case notification_data_pb2.SV_FAIL_REASON_DUPLICATE: 2007 ↛ 2009line 2007 didn't jump to line 2009 because the pattern on line 2007 always matched

2008 reason_string_key = ".reason_duplicate" 

2009 case _: 

2010 raise Exception("Shouldn't get here") 

2011 builder.para(reason_string_key) 

2012 return builder.build() 

2013 

2014 @classmethod 

2015 def from_notification(cls, data: notification_data_pb2.VerificationSVFail, *, user_name: str) -> Self: 

2016 return cls(user_name=user_name, reason=data.reason) 

2017 

2018 @classmethod 

2019 def test_instances(cls) -> list[Self]: 

2020 prototype = cls( 

2021 user_name="Alice", 

2022 reason=notification_data_pb2.SV_FAIL_REASON_NOT_A_PASSPORT, 

2023 ) 

2024 return [ 

2025 replace(prototype, reason=notification_data_pb2.SV_FAIL_REASON_WRONG_BIRTHDATE_OR_GENDER), 

2026 replace(prototype, reason=notification_data_pb2.SV_FAIL_REASON_NOT_A_PASSPORT), 

2027 replace(prototype, reason=notification_data_pb2.SV_FAIL_REASON_DUPLICATE), 

2028 ] 

2029 

2030 

2031@dataclass(kw_only=True, slots=True) 

2032class StrongVerificationSucceededEmail(EmailBase): 

2033 """Sent to a user when their strong verification has succeeded.""" 

2034 

2035 @property 

2036 def string_key_base(self) -> str: 

2037 return "strong_verification.succeeded" 

2038 

2039 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

2040 builder = self._body_builder(loc_context, security_warning=True) 

2041 builder.para(".success_message") 

2042 builder.para(".thanks_message") 

2043 builder.para(".cost_explanation") 

2044 builder.para(".donation_request") 

2045 donate_link = urls.donation_url() + "?utm_source=strong-verification-email" 

2046 builder.action(donate_link, ".donate_action") 

2047 return builder.build() 

2048 

2049 @classmethod 

2050 def test_instances(cls) -> list[Self]: 

2051 return [cls(user_name="Alice")] 

2052 

2053 

2054@dataclass(kw_only=True, slots=True) 

2055class ThreadReplyEmail(EmailBase): 

2056 """Sent to a user when someone replies in a comment thread they participated in.""" 

2057 

2058 author: UserInfo 

2059 parent_context: str # Title of the event or discussion being replied in 

2060 markdown_text: str 

2061 view_link: str 

2062 

2063 @property 

2064 def string_key_base(self) -> str: 

2065 return "thread_reply" 

2066 

2067 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

2068 return self._localize( 

2069 loc_context, ".subject", {"author": self.author.name, "parent_context": self.parent_context} 

2070 ) 

2071 

2072 def get_preview_line(self, loc_context: LocalizationContext) -> str | None: 

2073 return markdown_to_plaintext(self.markdown_text) 

2074 

2075 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

2076 builder = self._body_builder(loc_context) 

2077 builder.para(".body", {"author": self.author.name, "parent_context": self.parent_context}) 

2078 builder.user(self.author) 

2079 builder.quote(self.markdown_text, markdown=True) 

2080 builder.action(self.view_link, ".view_action") 

2081 return builder.build() 

2082 

2083 @classmethod 

2084 def from_notification(cls, data: notification_data_pb2.ThreadReply, *, user_name: str) -> Self: 

2085 parent = data.WhichOneof("reply_parent") 

2086 if parent == "event": 

2087 parent_context = data.event.title 

2088 view_link = urls.event_link(occurrence_id=data.event.event_id, slug=data.event.slug) 

2089 elif parent == "discussion": 2089 ↛ 2093line 2089 didn't jump to line 2093 because the condition on line 2089 was always true

2090 parent_context = data.discussion.title 

2091 view_link = urls.discussion_link(discussion_id=data.discussion.discussion_id, slug=data.discussion.slug) 

2092 else: 

2093 raise Exception("Can only do replies to events and discussions") 

2094 return cls( 

2095 user_name=user_name, 

2096 author=UserInfo.from_protobuf(data.author), 

2097 parent_context=parent_context, 

2098 markdown_text=data.reply.content, 

2099 view_link=view_link, 

2100 ) 

2101 

2102 @classmethod 

2103 def test_instances(cls) -> list[Self]: 

2104 return [ 

2105 cls( 

2106 user_name="Alice", 

2107 author=UserInfo.dummy_bob(), 

2108 parent_context="Best hiking trails near Berlin", 

2109 markdown_text="I agree, the Grünewald is **amazing**!", 

2110 view_link="https://couchers.org/discussions/123", 

2111 ) 

2112 ] 

2113 

2114 

2115def _localize_host_request_date(value: date, loc_context: LocalizationContext) -> str: 

2116 return loc_context.localize_date(value, with_year=False, with_day_of_week=True)