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

85 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 17:15 +0000

1from abc import abstractmethod 1a

2from datetime import datetime 1a

3from enum import StrEnum 1a

4from typing import Annotated, Literal 1a

5 

6from babel.numbers import format_currency 1a

7from pydantic import UUID4, BaseModel, Discriminator, computed_field 1a

8 

9from polar.email.react import render_email_template 1a

10from polar.kit.schemas import Schema 1a

11 

12 

13class NotificationType(StrEnum): 1a

14 maintainer_new_paid_subscription = "MaintainerNewPaidSubscriptionNotification" 1a

15 maintainer_new_product_sale = "MaintainerNewProductSaleNotification" 1a

16 maintainer_create_account = "MaintainerCreateAccountNotification" 1a

17 

18 

19class NotificationPayloadBase(BaseModel): 1a

20 @abstractmethod 1a

21 def subject(self) -> str: 1a

22 pass 

23 

24 @classmethod 1a

25 @abstractmethod 1a

26 def template_name(cls) -> str: 1a

27 pass 

28 

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

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): 1a

43 id: UUID4 1a

44 created_at: datetime 1a

45 type: NotificationType 1a

46 

47 

48class MaintainerAccountUnderReviewNotificationPayload(NotificationPayloadBase): 1a

49 account_type: str 1a

50 

51 def subject(self) -> str: 1a

52 return "Your Polar account is being reviewed" 

53 

54 @classmethod 1a

55 def template_name(cls) -> str: 1a

56 return "notification_account_under_review" 

57 

58 

59class MaintainerNewPaidSubscriptionNotificationPayload(NotificationPayloadBase): 1a

60 subscriber_name: str 1a

61 tier_name: str 1a

62 tier_price_amount: int | None 1a

63 tier_price_recurring_interval: str 1a

64 tier_organization_name: str 1a

65 

66 @computed_field 1a

67 def formatted_price_amount(self) -> str: 1a

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: 1a

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 1a

82 def template_name(cls) -> str: 1a

83 return "notification_new_subscription" 

84 

85 

86class MaintainerNewPaidSubscriptionNotification(NotificationBase): 1a

87 type: Literal[NotificationType.maintainer_new_paid_subscription] 1a

88 payload: MaintainerNewPaidSubscriptionNotificationPayload 1a

89 

90 

91class MaintainerNewProductSaleNotificationPayload(NotificationPayloadBase): 1a

92 customer_name: str 1a

93 product_name: str 1a

94 product_price_amount: int 1a

95 organization_name: str 1a

96 

97 @computed_field 1a

98 def formatted_price_amount(self) -> str: 1a

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

100 

101 def subject(self) -> str: 1a

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

103 

104 @classmethod 1a

105 def template_name(cls) -> str: 1a

106 return "notification_new_sale" 

107 

108 

109class MaintainerNewProductSaleNotification(NotificationBase): 1a

110 type: Literal[NotificationType.maintainer_new_product_sale] 1a

111 payload: MaintainerNewProductSaleNotificationPayload 1a

112 

113 

114class MaintainerCreateAccountNotificationPayload(NotificationPayloadBase): 1a

115 organization_name: str 1a

116 url: str 1a

117 

118 def subject(self) -> str: 1a

119 return ( 

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

121 ) 

122 

123 @classmethod 1a

124 def template_name(cls) -> str: 1a

125 return "notification_create_account" 

126 

127 

128class MaintainerCreateAccountNotification(NotificationBase): 1a

129 type: Literal[NotificationType.maintainer_create_account] 1a

130 payload: MaintainerCreateAccountNotificationPayload 1a

131 

132 

133NotificationPayload = ( 1a

134 MaintainerNewPaidSubscriptionNotificationPayload 

135 | MaintainerNewProductSaleNotificationPayload 

136 | MaintainerCreateAccountNotificationPayload 

137) 

138 

139Notification = Annotated[ 1a

140 MaintainerNewPaidSubscriptionNotification 

141 | MaintainerNewProductSaleNotification 

142 | MaintainerCreateAccountNotification, 

143 Discriminator(discriminator="type"), 

144]