Coverage for rfpy/web/hooks/mail_delivered.py: 100%
28 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-04-24 10:52 +0000
« prev ^ index » next coverage.py v7.8.0, created at 2025-04-24 10:52 +0000
1import logging
2from datetime import datetime
4from rfpy.model import EmailNotification
5from .webapp import WebhookApp, NotificationHook, PostmarkVars
7log = logging.getLogger(__name__)
10def postmark_isodate(datestring: str):
11 """
12 The microseconds portion of postmark api's datestring seems to
13 have 7, not 6, digits
14 2014-08-01T13:28:10.2735391-04:00
15 should be
16 2014-08-01T13:28:10.273539-04:00
17 """
18 if len(datestring) == 33:
19 datestring = datestring[0:26] + datestring[27:]
20 return datetime.fromisoformat(datestring)
23class DeliveredHook(NotificationHook):
24 new_status = EmailNotification.Status.delivered
26 def update_notification_record(self, data: PostmarkVars, en: EmailNotification):
27 try:
28 new_date = postmark_isodate(data.api_dict["DeliveredAt"])
29 except ValueError:
30 new_date = datetime.now()
32 en.delivered_date = new_date
35class BounceHook(NotificationHook):
36 new_status = EmailNotification.Status.bounced
38 def update_notification_record(self, data: PostmarkVars, en: EmailNotification):
39 bounce_type = data.api_dict.get("Name", "Bounce")
40 bounce_description = data.api_dict.get("Description", "Details not provided")
41 msg = f"{bounce_type}: {bounce_description} \nAt: {datetime.now()}"
42 if en.error is None:
43 en.error = msg
44 else:
45 en.error = en.error + "\n\n" + msg
48WebhookApp.register_route("/delivered", DeliveredHook())
49WebhookApp.register_route("/bounced", BounceHook())