Coverage for polar/notifications/notification.py: 66%

1from abc import abstractmethod 

2from datetime import datetime 

3from enum import StrEnum 

4from typing import Annotated, Literal 

5 

6from babel.numbers import format_currency 

7from pydantic import UUID4, BaseModel, Discriminator, computed_field 

8 

9from polar.email.react import render_email_template 

10from polar.kit.schemas import Schema 

11 

12 

13class NotificationType(StrEnum): 

14 maintainer_new_paid_subscription = "MaintainerNewPaidSubscriptionNotification" 

15 maintainer_new_product_sale = "MaintainerNewProductSaleNotification" 

16 maintainer_create_account = "MaintainerCreateAccountNotification" 

17 

18 

19class NotificationPayloadBase(BaseModel): 

20 @abstractmethod 

21 def subject(self) -> str: 

22 pass 

23 

24 @classmethod 

25 @abstractmethod 

26 def template_name(cls) -> str: 

27 pass 

28 

29 def render(self) -> tuple[str, str]: 

30 from polar.email.schemas import EmailAdapter 

31 

32 return self.subject(), render_email_template( 

33 EmailAdapter.validate_python( 

34 { 

35 "template": self.template_name(), 

36 "props": self, 

37 } 

38 ) 

39 ) 

40 

41 

42class NotificationBase(Schema): 

43 id: UUID4 

44 created_at: datetime 

45 type: NotificationType 

46 

47 

48class MaintainerAccountUnderReviewNotificationPayload(NotificationPayloadBase): 

49 account_type: str 

50 

51 def subject(self) -> str: 

52 return "Your Polar account is being reviewed" 

53 

54 @classmethod 

55 def template_name(cls) -> str: 

56 return "notification_account_under_review" 

57 

58 

59class MaintainerNewPaidSubscriptionNotificationPayload(NotificationPayloadBase): 

60 subscriber_name: str 

61 tier_name: str 

62 tier_price_amount: int | None 

63 tier_price_recurring_interval: str 

64 tier_organization_name: str 

65 

66 @computed_field 

67 def formatted_price_amount(self) -> str: 

68 if self.tier_price_amount is None: 

69 return "" 

70 return format_currency(self.tier_price_amount / 100, "USD", locale="en_US") 

71 

72 def subject(self) -> str: 

73 if self.tier_price_amount: 

74 price = ( 

75 f"{self.formatted_price_amount}/{self.tier_price_recurring_interval}" 

76 ) 

77 else: 

78 price = "free" 

79 return f"You have a new subscriber on {self.tier_name} ({price})!" 

80 

81 @classmethod 

82 def template_name(cls) -> str: 

83 return "notification_new_subscription" 

84 

85 

86class MaintainerNewPaidSubscriptionNotification(NotificationBase): 

87 type: Literal[NotificationType.maintainer_new_paid_subscription] 

88 payload: MaintainerNewPaidSubscriptionNotificationPayload 

89 

90 

91class MaintainerNewProductSaleNotificationPayload(NotificationPayloadBase): 

92 customer_name: str 

93 product_name: str 

94 product_price_amount: int 

95 organization_name: str 

96 

97 @computed_field 

98 def formatted_price_amount(self) -> str: 

99 return format_currency(self.product_price_amount / 100, "USD", locale="en_US") 

100 

101 def subject(self) -> str: 

102 return f"You've made a new sale ({self.formatted_price_amount})!" 

103 

104 @classmethod 

105 def template_name(cls) -> str: 

106 return "notification_new_sale" 

107 

108 

109class MaintainerNewProductSaleNotification(NotificationBase): 

110 type: Literal[NotificationType.maintainer_new_product_sale] 

111 payload: MaintainerNewProductSaleNotificationPayload 

112 

113 

114class MaintainerCreateAccountNotificationPayload(NotificationPayloadBase): 

115 organization_name: str 

116 url: str 

117 

118 def subject(self) -> str: 

119 return ( 

120 f"Create a payout account for {self.organization_name} now to receive funds" 

121 ) 

122 

123 @classmethod 

124 def template_name(cls) -> str: 

125 return "notification_create_account" 

126 

127 

128class MaintainerCreateAccountNotification(NotificationBase): 

129 type: Literal[NotificationType.maintainer_create_account] 

130 payload: MaintainerCreateAccountNotificationPayload 

131 

132 

133NotificationPayload = ( 

134 MaintainerNewPaidSubscriptionNotificationPayload 

135 | MaintainerNewProductSaleNotificationPayload 

136 | MaintainerCreateAccountNotificationPayload 

137) 

138 

139Notification = Annotated[ 

140 MaintainerNewPaidSubscriptionNotification 

141 | MaintainerNewProductSaleNotification 

142 | MaintainerCreateAccountNotification, 

143 Discriminator(discriminator="type"), 

144]