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

1085 statements  

« prev     ^ index     » next       coverage.py v7.14.3, created at 2026-06-29 22:41 +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 address: str | None # The None case handles legacy online events 

616 view_url: str 

617 description_markdown: str 

618 

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

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

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

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

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

624 

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

626 html += "<br>" 

627 html += time_range_display 

628 if self.address: 

629 html += "<br>" 

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

631 

632 return ParaBlock(text=Markup(html)) 

633 

634 def get_description_block(self) -> EmailBlock: 

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

636 

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

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

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

640 

641 @classmethod 

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

643 return cls( 

644 title=event.title, 

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

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

647 # Backcompat (2026-06): We might still have queued notifications referencing events with online_information. 

648 address=(event.location.address or None) if event.HasField("location") else None, 

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

650 description_markdown=event.content or "", 

651 ) 

652 

653 @staticmethod 

654 def dummy() -> EventInfo: 

655 return EventInfo( 

656 title="Berlin Meetup", 

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

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

659 address="Alexanderplatz, Berlin", 

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

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

662 ) 

663 

664 

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

666class EventCreatedEmail(EmailBase): 

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

668 

669 inviting_user: UserInfo 

670 event_info: EventInfo 

671 community_name: str | None 

672 community_url: str | None 

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

674 

675 @property 

676 def string_key_base(self) -> str: 

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

678 

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

680 return self._localize( 

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

682 ) 

683 

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

685 return markdown_to_plaintext(self.event_info.description_markdown) 

686 

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

688 builder = self._body_builder(loc_context) 

689 if self.community_name: 

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

691 else: 

692 builder.para(".body_no_community") 

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

694 builder.user(self.inviting_user) 

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

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

697 return builder.build() 

698 

699 @classmethod 

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

701 has_community = bool(data.in_community.community_id) 

702 community_url = ( 

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

704 if has_community 

705 else None 

706 ) 

707 return cls( 

708 user_name=user_name, 

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

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

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

712 community_url=community_url, 

713 is_invite=is_invite, 

714 ) 

715 

716 @classmethod 

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

718 prototype = cls( 

719 user_name="Alice", 

720 inviting_user=UserInfo.dummy_bob(), 

721 event_info=EventInfo.dummy(), 

722 community_name="Berlin", 

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

724 is_invite=True, 

725 ) 

726 return [ 

727 replace(prototype, is_invite=True), 

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

729 replace(prototype, is_invite=False), 

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

731 ] 

732 

733 

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

735class EventUpdatedEmail(EmailBase): 

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

737 

738 updating_user: UserInfo 

739 event_info: EventInfo 

740 updated_items: list[notification_data_pb2.EventUpdateItem.ValueType] 

741 

742 @property 

743 def string_key_base(self) -> str: 

744 return "events.updated" 

745 

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

747 return self._localize( 

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

749 ) 

750 

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

752 builder = self._body_builder(loc_context) 

753 builder.para(".body") 

754 

755 updated_items_string_keys = list( 

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

757 ) 

758 if updated_items_string_keys: 

759 updated_items_text = loc_context.localize_list( 

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

761 ) 

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

763 else: 

764 builder.para(".updated_generic") 

765 

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

767 builder.user(self.updating_user) 

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

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

770 return builder.build() 

771 

772 @classmethod 

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

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

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

776 updated_items.extend(data.updated_enum_items) 

777 elif data.updated_str_items: 

778 for updated_str_item in data.updated_str_items: 

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

780 updated_items.append(updated_enum_item) 

781 

782 return cls( 

783 user_name=user_name, 

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

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

786 updated_items=updated_items, 

787 ) 

788 

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

790 @staticmethod 

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

792 match value: 

793 case "title": 

794 return notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_TITLE 

795 case "content": 

796 return notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_CONTENT 

797 case "location": 

798 return notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_LOCATION 

799 case "start time": 

800 return notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_START_TIME 

801 case "end time": 

802 return notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_END_TIME 

803 case _: 

804 return None 

805 

806 @staticmethod 

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

808 match value: 

809 case notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_TITLE: 

810 return ".item_names.title" 

811 case notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_CONTENT: 

812 return ".item_names.content" 

813 case notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_LOCATION: 

814 return ".item_names.location" 

815 case notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_START_TIME: 

816 return ".item_names.start_time" 

817 case notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_END_TIME: 

818 return ".item_names.end_time" 

819 case _: 

820 return None 

821 

822 @classmethod 

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

824 prototype = cls( 

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

826 ) 

827 return [ 

828 replace(prototype, updated_items=[]), 

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

830 ] 

831 

832 

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

834class EventOrganizerInvitedEmail(EmailBase): 

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

836 

837 inviting_user: UserInfo 

838 event_info: EventInfo 

839 

840 @property 

841 def string_key_base(self) -> str: 

842 return "events.organizer_invited" 

843 

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

845 return self._localize( 

846 loc_context, 

847 ".subject", 

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

849 ) 

850 

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

852 builder = self._body_builder(loc_context) 

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

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

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

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

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

858 builder.para(_do_not_reply_request_string_key) 

859 return builder.build() 

860 

861 @classmethod 

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

863 return cls( 

864 user_name=user_name, 

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

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

867 ) 

868 

869 @classmethod 

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

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

872 

873 

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

875class EventCommentEmail(EmailBase): 

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

877 

878 author: UserInfo 

879 event_info: EventInfo 

880 comment_markdown: str 

881 

882 @property 

883 def string_key_base(self) -> str: 

884 return "events.comment" 

885 

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

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

888 

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

890 return markdown_to_plaintext(self.comment_markdown) 

891 

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

893 builder = self._body_builder(loc_context) 

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

895 builder.user(self.author) 

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

897 builder.para(".event_details") 

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

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

900 return builder.build() 

901 

902 @classmethod 

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

904 return cls( 

905 user_name=user_name, 

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

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

908 comment_markdown=data.reply.content, 

909 ) 

910 

911 @classmethod 

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

913 return [ 

914 cls( 

915 user_name="Alice", 

916 author=UserInfo.dummy_bob(), 

917 event_info=EventInfo.dummy(), 

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

919 ) 

920 ] 

921 

922 

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

924class EventReminderEmail(EmailBase): 

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

926 

927 event_info: EventInfo 

928 

929 @property 

930 def string_key_base(self) -> str: 

931 return "events.reminder" 

932 

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

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

935 

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

937 builder = self._body_builder(loc_context) 

938 builder.para(".body") 

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

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

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

942 return builder.build() 

943 

944 @classmethod 

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

946 return cls( 

947 user_name=user_name, 

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

949 ) 

950 

951 @classmethod 

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

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

954 

955 

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

957class EventCancelledEmail(EmailBase): 

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

959 

960 cancelling_user: UserInfo 

961 event_info: EventInfo 

962 

963 @property 

964 def string_key_base(self) -> str: 

965 return "events.cancel" 

966 

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

968 return self._localize( 

969 loc_context, 

970 ".subject", 

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

972 ) 

973 

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

975 builder = self._body_builder(loc_context) 

976 builder.para(".body") 

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

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

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

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

981 return builder.build() 

982 

983 @classmethod 

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

985 return cls( 

986 user_name=user_name, 

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

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

989 ) 

990 

991 @classmethod 

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

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

994 

995 

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

997class EventDeletedEmail(EmailBase): 

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

999 

1000 event_info: EventInfo 

1001 

1002 @property 

1003 def string_key_base(self) -> str: 

1004 return "events.deleted" 

1005 

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

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

1008 

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

1010 builder = self._body_builder(loc_context) 

1011 builder.para(".body") 

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

1013 return builder.build() 

1014 

1015 @classmethod 

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

1017 return cls( 

1018 user_name=user_name, 

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

1020 ) 

1021 

1022 @classmethod 

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

1024 return [ 

1025 cls( 

1026 user_name="Alice", 

1027 event_info=EventInfo.dummy(), 

1028 ) 

1029 ] 

1030 

1031 

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

1033class FriendReferenceReceivedEmail(EmailBase): 

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

1035 

1036 from_user: UserInfo 

1037 text: str 

1038 

1039 @property 

1040 def string_key_base(self) -> str: 

1041 return "references.received.friend" 

1042 

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

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

1045 

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

1047 return self.text 

1048 

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

1050 builder = self._body_builder(loc_context) 

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

1052 builder.user(self.from_user) 

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

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

1055 return builder.build() 

1056 

1057 @classmethod 

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

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

1060 

1061 @classmethod 

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

1063 return [ 

1064 cls( 

1065 user_name="Alice", 

1066 from_user=UserInfo.dummy_bob(), 

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

1068 ) 

1069 ] 

1070 

1071 

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

1073class FriendRequestReceivedEmail(EmailBase): 

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

1075 

1076 befriender: UserInfo 

1077 

1078 @property 

1079 def string_key_base(self) -> str: 

1080 return "friend_requests.received" 

1081 

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

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

1084 

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

1086 builder = self._body_builder(loc_context) 

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

1088 builder.user(self.befriender) 

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

1090 builder.para(".closing") 

1091 builder.para(_do_not_reply_request_string_key) 

1092 return builder.build() 

1093 

1094 @classmethod 

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

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

1097 

1098 @classmethod 

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

1100 return [ 

1101 cls( 

1102 user_name="Alice", 

1103 befriender=UserInfo.dummy_bob(), 

1104 ) 

1105 ] 

1106 

1107 

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

1109class FriendRequestAcceptedEmail(EmailBase): 

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

1111 

1112 new_friend: UserInfo 

1113 

1114 @property 

1115 def string_key_base(self) -> str: 

1116 return "friend_requests.accepted" 

1117 

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

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

1120 

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

1122 builder = self._body_builder(loc_context) 

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

1124 builder.user(self.new_friend) 

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

1126 builder.para(".closing") 

1127 return builder.build() 

1128 

1129 @classmethod 

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

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

1132 

1133 @classmethod 

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

1135 return [ 

1136 cls( 

1137 user_name="Alice", 

1138 new_friend=UserInfo.dummy_bob(), 

1139 ) 

1140 ] 

1141 

1142 

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

1144class GenderChangedEmail(EmailBase): 

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

1146 

1147 new_gender: str 

1148 

1149 @property 

1150 def string_key_base(self) -> str: 

1151 return "gender_changed" 

1152 

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

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

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

1156 return builder.build() 

1157 

1158 @classmethod 

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

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

1161 

1162 @classmethod 

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

1164 return [ 

1165 cls( 

1166 user_name="Alice", 

1167 new_gender="Male", 

1168 ) 

1169 ] 

1170 

1171 

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

1173class HostRequestCreatedEmail(EmailBase): 

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

1175 

1176 surfer: UserInfo 

1177 from_date: date 

1178 to_date: date 

1179 text: str 

1180 quick_decline_link: str 

1181 view_link: str 

1182 

1183 @property 

1184 def string_key_base(self) -> str: 

1185 return "host_requests.created" 

1186 

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

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

1189 

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

1191 return self.text 

1192 

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

1194 builder = self._body_builder(loc_context) 

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

1196 builder.user( 

1197 self.surfer, 

1198 "host_requests.generic.date_range", 

1199 { 

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

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

1202 }, 

1203 ) 

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

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

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

1207 builder.para(".respond_encouragement") 

1208 builder.para(_do_not_reply_request_string_key) 

1209 return builder.build() 

1210 

1211 @classmethod 

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

1213 return cls( 

1214 user_name, 

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

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

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

1218 text=data.text, 

1219 quick_decline_link=generate_quick_decline_link(data.host_request), 

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

1221 ) 

1222 

1223 @classmethod 

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

1225 return [ 

1226 cls( 

1227 user_name="Alice", 

1228 surfer=UserInfo.dummy_bob(), 

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

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

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

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

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

1234 ) 

1235 ] 

1236 

1237 

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

1239class HostRequestReminderEmail(EmailBase): 

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

1241 

1242 surfer: UserInfo 

1243 from_date: date 

1244 to_date: date 

1245 view_link: str 

1246 

1247 @property 

1248 def string_key_base(self) -> str: 

1249 return "host_requests.reminder" 

1250 

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

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

1253 

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

1255 builder = self._body_builder(loc_context) 

1256 builder.para(".body") 

1257 builder.user( 

1258 self.surfer, 

1259 "host_requests.generic.date_range", 

1260 { 

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

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

1263 }, 

1264 ) 

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

1266 builder.para(_do_not_reply_request_string_key) 

1267 return builder.build() 

1268 

1269 @classmethod 

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

1271 return cls( 

1272 user_name, 

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

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

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

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

1277 ) 

1278 

1279 @classmethod 

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

1281 return [ 

1282 cls( 

1283 user_name="Alice", 

1284 surfer=UserInfo.dummy_bob(), 

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

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

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

1288 ) 

1289 ] 

1290 

1291 

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

1293class HostRequestMessageEmail(EmailBase): 

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

1295 

1296 other_user: UserInfo 

1297 from_date: date 

1298 to_date: date 

1299 text: str 

1300 from_host: bool 

1301 view_link: str 

1302 

1303 @property 

1304 def string_key_base(self) -> str: 

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

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

1307 

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

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

1310 

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

1312 return self.text 

1313 

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

1315 builder = self._body_builder(loc_context) 

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

1317 builder.user( 

1318 self.other_user, 

1319 "host_requests.generic.date_range", 

1320 { 

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

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

1323 }, 

1324 ) 

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

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

1327 builder.para(_do_not_reply_request_string_key) 

1328 return builder.build() 

1329 

1330 @classmethod 

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

1332 return cls( 

1333 user_name, 

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

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

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

1337 text=data.text, 

1338 from_host=not data.am_host, 

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

1340 ) 

1341 

1342 @classmethod 

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

1344 prototype = cls( 

1345 user_name="Alice", 

1346 other_user=UserInfo.dummy_bob(), 

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

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

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

1350 from_host=True, 

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

1352 ) 

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

1354 

1355 

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

1357class HostRequestMissedMessagesEmail(EmailBase): 

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

1359 

1360 other_user: UserInfo 

1361 from_date: date 

1362 to_date: date 

1363 from_host: bool 

1364 view_link: str 

1365 

1366 @property 

1367 def string_key_base(self) -> str: 

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

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

1370 

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

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

1373 

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

1375 builder = self._body_builder(loc_context) 

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

1377 builder.user( 

1378 self.other_user, 

1379 "host_requests.generic.date_range", 

1380 { 

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

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

1383 }, 

1384 ) 

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

1386 builder.para(_do_not_reply_request_string_key) 

1387 return builder.build() 

1388 

1389 @classmethod 

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

1391 return cls( 

1392 user_name, 

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

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

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

1396 from_host=not data.am_host, 

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

1398 ) 

1399 

1400 @classmethod 

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

1402 prototype = cls( 

1403 user_name="Alice", 

1404 other_user=UserInfo.dummy_bob(), 

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

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

1407 from_host=True, 

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

1409 ) 

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

1411 

1412 

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

1414class HostRequestStatusChangedEmail(EmailBase): 

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

1416 

1417 other_user: UserInfo 

1418 from_date: date 

1419 to_date: date 

1420 new_status: conversations_pb2.HostRequestStatus.ValueType 

1421 view_link: str 

1422 

1423 @property 

1424 def string_key_base(self) -> str: 

1425 base_key = "host_requests.status_changed" 

1426 match self.new_status: 

1427 case conversations_pb2.HOST_REQUEST_STATUS_ACCEPTED: 

1428 return f"{base_key}.accepted_by_host" 

1429 case conversations_pb2.HOST_REQUEST_STATUS_REJECTED: 

1430 return f"{base_key}.declined_by_host" 

1431 case conversations_pb2.HOST_REQUEST_STATUS_CONFIRMED: 

1432 return f"{base_key}.confirmed_by_surfer" 

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

1434 return f"{base_key}.cancelled_by_surfer" 

1435 case _: 

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

1437 

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

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

1440 

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

1442 builder = self._body_builder(loc_context) 

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

1444 builder.user( 

1445 self.other_user, 

1446 "host_requests.generic.date_range", 

1447 { 

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

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

1450 }, 

1451 ) 

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

1453 builder.para(_do_not_reply_request_string_key) 

1454 return builder.build() 

1455 

1456 @classmethod 

1457 def from_notification( 

1458 cls, 

1459 data: notification_data_pb2.HostRequestAccept 

1460 | notification_data_pb2.HostRequestReject 

1461 | notification_data_pb2.HostRequestConfirm 

1462 | notification_data_pb2.HostRequestCancel, 

1463 *, 

1464 user_name: str, 

1465 ) -> Self: 

1466 other_user: UserInfo 

1467 new_status: conversations_pb2.HostRequestStatus.ValueType 

1468 match data: 

1469 case notification_data_pb2.HostRequestAccept(): 

1470 other_user = UserInfo.from_protobuf(data.host) 

1471 new_status = conversations_pb2.HostRequestStatus.HOST_REQUEST_STATUS_ACCEPTED 

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

1473 other_user = UserInfo.from_protobuf(data.host) 

1474 new_status = conversations_pb2.HostRequestStatus.HOST_REQUEST_STATUS_REJECTED 

1475 case notification_data_pb2.HostRequestConfirm(): 

1476 other_user = UserInfo.from_protobuf(data.surfer) 

1477 new_status = conversations_pb2.HostRequestStatus.HOST_REQUEST_STATUS_CONFIRMED 

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

1479 other_user = UserInfo.from_protobuf(data.surfer) 

1480 new_status = conversations_pb2.HostRequestStatus.HOST_REQUEST_STATUS_CANCELLED 

1481 case _: 

1482 # Enable mypy's exhaustiveness checking 

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

1484 

1485 return cls( 

1486 user_name, 

1487 other_user=other_user, 

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

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

1490 new_status=new_status, 

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

1492 ) 

1493 

1494 @classmethod 

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

1496 prototype = cls( 

1497 user_name="Alice", 

1498 other_user=UserInfo.dummy_bob(), 

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

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

1501 new_status=conversations_pb2.HOST_REQUEST_STATUS_ACCEPTED, 

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

1503 ) 

1504 return [ 

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

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

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

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

1509 ] 

1510 

1511 

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

1513class HostReferenceReceivedEmail(EmailBase): 

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

1515 

1516 from_user: UserInfo 

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

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

1519 leave_reference_url: str 

1520 

1521 @property 

1522 def string_key_base(self) -> str: 

1523 return "references.received" 

1524 

1525 @property 

1526 def string_role_subkey(self) -> str: 

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

1528 

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

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

1531 

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

1533 return self.text 

1534 

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

1536 builder = self._body_builder(loc_context) 

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

1538 builder.user(self.from_user) 

1539 if self.text: 

1540 builder.para(".before_quote") 

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

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

1543 else: 

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

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

1546 return builder.build() 

1547 

1548 @classmethod 

1549 def from_notification( 

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

1551 ) -> Self: 

1552 return cls( 

1553 user_name=user_name, 

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

1555 text=data.text or None, 

1556 surfed=surfed, 

1557 leave_reference_url=urls.leave_reference_link( 

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

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

1560 host_request_id=str(data.host_request_id), 

1561 ), 

1562 ) 

1563 

1564 @classmethod 

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

1566 prototype = cls( 

1567 user_name="Alice", 

1568 from_user=UserInfo.dummy_bob(), 

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

1570 surfed=True, 

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

1572 ) 

1573 return [ 

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

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

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

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

1578 ] 

1579 

1580 

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

1582class HostReferenceReminderEmail(EmailBase): 

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

1584 

1585 other_user: UserInfo 

1586 days_left: int 

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

1588 leave_reference_url: str 

1589 

1590 @property 

1591 def string_key_base(self) -> str: 

1592 return "references.reminder" 

1593 

1594 @property 

1595 def string_role_subkey(self) -> str: 

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

1597 

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

1599 return self._localize( 

1600 loc_context, 

1601 ".subject_days", 

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

1603 ) 

1604 

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

1606 builder = self._body_builder(loc_context) 

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

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

1609 builder.user(self.other_user) 

1610 builder.action( 

1611 self.leave_reference_url, 

1612 "references.write_action", 

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

1614 ) 

1615 builder.para(".via_messaging_note") 

1616 builder.para(".importance_note") 

1617 builder.para(".visibility_note") 

1618 return builder.build() 

1619 

1620 @classmethod 

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

1622 return cls( 

1623 user_name=user_name, 

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

1625 days_left=data.days_left, 

1626 surfed=surfed, 

1627 leave_reference_url=urls.leave_reference_link( 

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

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

1630 host_request_id=str(data.host_request_id), 

1631 ), 

1632 ) 

1633 

1634 @classmethod 

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

1636 prototype = cls( 

1637 user_name="Alice", 

1638 other_user=UserInfo.dummy_bob(), 

1639 days_left=7, 

1640 surfed=True, 

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

1642 ) 

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

1644 

1645 

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

1647class ModeratorNoteEmail(EmailBase): 

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

1649 

1650 @property 

1651 def string_key_base(self) -> str: 

1652 return "moderator_note" 

1653 

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

1655 builder = self._body_builder(loc_context) 

1656 builder.para(".body") 

1657 return builder.build() 

1658 

1659 @classmethod 

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

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

1662 

1663 

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

1665class NewBlogPostEmail(EmailBase): 

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

1667 

1668 title: str 

1669 blurb: str 

1670 url: str 

1671 

1672 @property 

1673 def string_key_base(self) -> str: 

1674 return "new_blog_post" 

1675 

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

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

1678 

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

1680 return self.blurb 

1681 

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

1683 builder = self._body_builder(loc_context) 

1684 builder.para(".intro") 

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

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

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

1688 return builder.build() 

1689 

1690 @classmethod 

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

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

1693 

1694 @classmethod 

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

1696 return [ 

1697 cls( 

1698 user_name="Alice", 

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

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

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

1702 ) 

1703 ] 

1704 

1705 

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

1707class OnboardingReminderEmail(EmailBase): 

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

1709 

1710 initial: bool 

1711 

1712 @property 

1713 def string_key_base(self) -> str: 

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

1715 

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

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

1718 edit_profile_url = urls.edit_profile_link() 

1719 if self.initial: 

1720 builder.para(".welcome") 

1721 builder.para(".early_user_role") 

1722 builder.para(".fill_in_profile") 

1723 builder.para(".edit_profile_prompt") 

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

1725 builder.para(".share_with_friends") 

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

1727 builder.para(".platform_under_development") 

1728 builder.para(".thanks_for_joining") 

1729 builder.para(".signature") 

1730 else: 

1731 builder.para(".intro") 

1732 builder.para(".fill_in_profile") 

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

1734 builder.para(".no_empty_accounts") 

1735 builder.para(".profile_importance") 

1736 builder.para(".signature") 

1737 return builder.build() 

1738 

1739 @classmethod 

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

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

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

1743 

1744 

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

1746class PasswordChangedEmail(EmailBase): 

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

1748 

1749 @property 

1750 def string_key_base(self) -> str: 

1751 return "password_changed" 

1752 

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

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

1755 builder.para(".body") 

1756 return builder.build() 

1757 

1758 @classmethod 

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

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

1761 

1762 

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

1764class PasswordResetCompletedEmail(EmailBase): 

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

1766 

1767 @property 

1768 def string_key_base(self) -> str: 

1769 return "password_reset.completed" 

1770 

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

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

1773 builder.para(".body") 

1774 return builder.build() 

1775 

1776 @classmethod 

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

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

1779 

1780 

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

1782class PasswordResetStartedEmail(EmailBase): 

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

1784 

1785 password_reset_link: str 

1786 

1787 @property 

1788 def string_key_base(self) -> str: 

1789 return "password_reset.started" 

1790 

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

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

1793 builder.para(".request_description") 

1794 builder.para(".confirmation_instructions") 

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

1796 return builder.build() 

1797 

1798 @classmethod 

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

1800 return cls( 

1801 user_name=user_name, 

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

1803 ) 

1804 

1805 @classmethod 

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

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

1808 

1809 

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

1811class PhoneNumberChangeEmail(EmailBase): 

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

1813 

1814 new_phone_number: str 

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

1816 

1817 @property 

1818 def string_key_base(self) -> str: 

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

1820 

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

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

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

1824 return builder.build() 

1825 

1826 @classmethod 

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

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

1829 

1830 @classmethod 

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

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

1833 

1834 @classmethod 

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

1836 prototype = cls( 

1837 user_name="Alice", 

1838 new_phone_number="+12223334444", 

1839 completed=False, 

1840 ) 

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

1842 

1843 

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

1845class PostalVerificationFailedEmail(EmailBase): 

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

1847 

1848 reason: notification_data_pb2.PostalVerificationFailReason.ValueType 

1849 

1850 @property 

1851 def string_key_base(self) -> str: 

1852 return "postal_verification.failed" 

1853 

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

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

1856 match self.reason: 

1857 case notification_data_pb2.POSTAL_VERIFICATION_FAIL_REASON_CODE_EXPIRED: 

1858 reason_string_key = ".reason_code_expired" 

1859 case notification_data_pb2.POSTAL_VERIFICATION_FAIL_REASON_TOO_MANY_ATTEMPTS: 

1860 reason_string_key = ".reason_too_many_attempts" 

1861 case _: 

1862 reason_string_key = ".reason_unknown" 

1863 builder.para(reason_string_key) 

1864 return builder.build() 

1865 

1866 @classmethod 

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

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

1869 

1870 @classmethod 

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

1872 prototype = cls( 

1873 user_name="Alice", 

1874 reason=notification_data_pb2.POSTAL_VERIFICATION_FAIL_REASON_CODE_EXPIRED, 

1875 ) 

1876 return [ 

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

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

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

1880 ] 

1881 

1882 

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

1884class PostalVerificationPostcardSentEmail(EmailBase): 

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

1886 

1887 city: str 

1888 country: str 

1889 

1890 @property 

1891 def string_key_base(self) -> str: 

1892 return "postal_verification.postcard_sent" 

1893 

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

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

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

1897 return builder.build() 

1898 

1899 @classmethod 

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

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

1902 

1903 @classmethod 

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

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

1906 

1907 

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

1909class PostalVerificationSucceededEmail(EmailBase): 

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

1911 

1912 @property 

1913 def string_key_base(self) -> str: 

1914 return "postal_verification.succeeded" 

1915 

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

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

1918 builder.para(".body") 

1919 return builder.build() 

1920 

1921 @classmethod 

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

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

1924 

1925 

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

1927class SignupVerifyEmail(EmailBase): 

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

1929 

1930 verify_url: str 

1931 

1932 @property 

1933 def string_key_base(self) -> str: 

1934 return "signup.verify" 

1935 

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

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

1938 

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

1940 builder = self._body_builder(loc_context) 

1941 builder.para(".thanks") 

1942 builder.para(".instructions") 

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

1944 builder.para("signup.closing") 

1945 return builder.build() 

1946 

1947 @classmethod 

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

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

1950 

1951 

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

1953class SignupContinueEmail(EmailBase): 

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

1955 

1956 continue_url: str 

1957 

1958 @property 

1959 def string_key_base(self) -> str: 

1960 return "signup.continue" 

1961 

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

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

1964 

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

1966 builder = self._body_builder(loc_context) 

1967 builder.para(".request") 

1968 builder.para(".instructions") 

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

1970 builder.para("signup.closing") 

1971 builder.para(".ignore_if_unexpected") 

1972 return builder.build() 

1973 

1974 @classmethod 

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

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

1977 

1978 

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

1980class StrongVerificationFailedEmail(EmailBase): 

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

1982 

1983 reason: notification_data_pb2.SVFailReason.ValueType 

1984 

1985 @property 

1986 def string_key_base(self) -> str: 

1987 return "strong_verification.failed" 

1988 

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

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

1991 match self.reason: 

1992 case notification_data_pb2.SV_FAIL_REASON_WRONG_BIRTHDATE_OR_GENDER: 

1993 reason_string_key = ".reason_wrong_birthdate_or_gender" 

1994 case notification_data_pb2.SV_FAIL_REASON_NOT_A_PASSPORT: 

1995 reason_string_key = ".reason_not_a_passport" 

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

1997 reason_string_key = ".reason_duplicate" 

1998 case _: 

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

2000 builder.para(reason_string_key) 

2001 return builder.build() 

2002 

2003 @classmethod 

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

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

2006 

2007 @classmethod 

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

2009 prototype = cls( 

2010 user_name="Alice", 

2011 reason=notification_data_pb2.SV_FAIL_REASON_NOT_A_PASSPORT, 

2012 ) 

2013 return [ 

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

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

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

2017 ] 

2018 

2019 

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

2021class StrongVerificationSucceededEmail(EmailBase): 

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

2023 

2024 @property 

2025 def string_key_base(self) -> str: 

2026 return "strong_verification.succeeded" 

2027 

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

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

2030 builder.para(".success_message") 

2031 builder.para(".thanks_message") 

2032 builder.para(".cost_explanation") 

2033 builder.para(".donation_request") 

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

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

2036 return builder.build() 

2037 

2038 @classmethod 

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

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

2041 

2042 

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

2044class ThreadReplyEmail(EmailBase): 

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

2046 

2047 author: UserInfo 

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

2049 markdown_text: str 

2050 view_link: str 

2051 

2052 @property 

2053 def string_key_base(self) -> str: 

2054 return "thread_reply" 

2055 

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

2057 return self._localize( 

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

2059 ) 

2060 

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

2062 return markdown_to_plaintext(self.markdown_text) 

2063 

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

2065 builder = self._body_builder(loc_context) 

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

2067 builder.user(self.author) 

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

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

2070 return builder.build() 

2071 

2072 @classmethod 

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

2074 parent = data.WhichOneof("reply_parent") 

2075 if parent == "event": 

2076 parent_context = data.event.title 

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

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

2079 parent_context = data.discussion.title 

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

2081 else: 

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

2083 return cls( 

2084 user_name=user_name, 

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

2086 parent_context=parent_context, 

2087 markdown_text=data.reply.content, 

2088 view_link=view_link, 

2089 ) 

2090 

2091 @classmethod 

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

2093 return [ 

2094 cls( 

2095 user_name="Alice", 

2096 author=UserInfo.dummy_bob(), 

2097 parent_context="Best hiking trails near Berlin", 

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

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

2100 ) 

2101 ] 

2102 

2103 

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

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