Coverage for polar/models/product.py: 78%

1from enum import StrEnum 

2from typing import TYPE_CHECKING 

3from uuid import UUID 

4 

5from sqlalchemy import ( 

6 Boolean, 

7 ColumnElement, 

8 ForeignKey, 

9 Integer, 

10 String, 

11 Text, 

12 Uuid, 

13 case, 

14 or_, 

15 select, 

16) 

17from sqlalchemy.dialects.postgresql import CITEXT 

18from sqlalchemy.ext.associationproxy import AssociationProxy, association_proxy 

19from sqlalchemy.ext.hybrid import hybrid_property 

20from sqlalchemy.orm import Mapped, declared_attr, mapped_column, relationship 

21 

22from polar.enums import SubscriptionRecurringInterval 

23from polar.kit.db.models import RecordModel 

24from polar.kit.extensions.sqlalchemy import StringEnum 

25from polar.kit.metadata import MetadataMixin 

26from polar.kit.tax import TaxCode 

27from polar.kit.trial import TrialConfigurationMixin 

28from polar.models.product_price import ProductPriceAmountType, ProductPriceType 

29 

30from .product_price import ProductPrice 

31 

32if TYPE_CHECKING: 32 ↛ 33line 32 didn't jump to line 33 because the condition on line 32 was never true

33 from polar.models import ( 

34 Benefit, 

35 Organization, 

36 ProductBenefit, 

37 ProductCustomField, 

38 ProductMedia, 

39 ) 

40 from polar.models.file import ProductMediaFile 

41 

42 

43class ProductBillingType(StrEnum): 

44 one_time = "one_time" 

45 recurring = "recurring" 

46 

47 

48class Product(TrialConfigurationMixin, MetadataMixin, RecordModel): 

49 __tablename__ = "products" 

50 

51 name: Mapped[str] = mapped_column(CITEXT(), nullable=False) 

52 description: Mapped[str | None] = mapped_column(Text, nullable=True) 

53 is_archived: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) 

54 recurring_interval: Mapped[SubscriptionRecurringInterval | None] = mapped_column( 

55 StringEnum(SubscriptionRecurringInterval), 

56 nullable=True, 

57 index=True, 

58 default=None, 

59 ) 

60 recurring_interval_count: Mapped[int | None] = mapped_column( 

61 Integer, nullable=True, default=None 

62 ) 

63 

64 is_tax_applicable: Mapped[bool] = mapped_column( 

65 Boolean, nullable=False, default=True 

66 ) 

67 tax_code: Mapped[TaxCode] = mapped_column( 

68 StringEnum(TaxCode), 

69 nullable=False, 

70 default=TaxCode.general_electronically_supplied_services, 

71 ) 

72 

73 stripe_product_id: Mapped[str | None] = mapped_column( 

74 String, nullable=True, index=True 

75 ) 

76 

77 organization_id: Mapped[UUID] = mapped_column( 

78 Uuid, 

79 ForeignKey("organizations.id", ondelete="cascade"), 

80 nullable=False, 

81 index=True, 

82 ) 

83 

84 @declared_attr 

85 def organization(cls) -> Mapped["Organization"]: 

86 return relationship("Organization", lazy="raise", back_populates="products") 

87 

88 @declared_attr 

89 def all_prices(cls) -> Mapped[list["ProductPrice"]]: 

90 return relationship( 

91 "ProductPrice", lazy="raise", cascade="all", back_populates="product" 

92 ) 

93 

94 @declared_attr 

95 def prices(cls) -> Mapped[list["ProductPrice"]]: 

96 # Prices are almost always needed, so eager loading makes sense 

97 return relationship( 

98 "ProductPrice", 

99 lazy="selectin", 

100 primaryjoin=( 

101 "and_(" 

102 "ProductPrice.product_id == Product.id, " 

103 "ProductPrice.is_archived.is_(False), " 

104 "ProductPrice.source == 'catalog'" 

105 ")" 

106 ), 

107 order_by="(case(" 

108 "(ProductPrice.amount_type.in_(['fixed', 'custom', 'free']), 0), " 

109 "(ProductPrice.amount_type == 'metered_unit', 1), " 

110 "), ProductPrice.created_at)", 

111 viewonly=True, 

112 ) 

113 

114 product_benefits: Mapped[list["ProductBenefit"]] = relationship( 

115 # Benefits are almost always needed, so eager loading makes sense 

116 lazy="selectin", 

117 order_by="ProductBenefit.order", 

118 cascade="all, delete-orphan", 

119 back_populates="product", 

120 ) 

121 

122 benefits: AssociationProxy[list["Benefit"]] = association_proxy( 

123 "product_benefits", "benefit" 

124 ) 

125 

126 product_medias: Mapped[list["ProductMedia"]] = relationship( 

127 lazy="raise", 

128 order_by="ProductMedia.order", 

129 cascade="all, delete-orphan", 

130 back_populates="product", 

131 ) 

132 

133 medias: AssociationProxy[list["ProductMediaFile"]] = association_proxy( 

134 "product_medias", "file" 

135 ) 

136 

137 attached_custom_fields: Mapped[list["ProductCustomField"]] = relationship( 

138 lazy="raise", 

139 order_by="ProductCustomField.order", 

140 cascade="all, delete-orphan", 

141 back_populates="product", 

142 ) 

143 

144 def get_stripe_name(self) -> str: 

145 return f"{self.organization.slug} - {self.name}" 

146 

147 def get_price( 

148 self, id: UUID, *, include_archived: bool = False 

149 ) -> "ProductPrice | None": 

150 prices = self.all_prices if include_archived else self.prices 

151 for price in prices: 

152 if price.id == id: 

153 return price 

154 return None 

155 

156 def get_static_price( 

157 self, *, include_archived: bool = False 

158 ) -> "ProductPrice | None": 

159 prices = self.all_prices if include_archived else self.prices 

160 for price in prices: 160 ↛ 163line 160 didn't jump to line 163 because the loop on line 160 didn't complete

161 if price.is_static: 161 ↛ 160line 161 didn't jump to line 160 because the condition on line 161 was always true

162 return price 

163 return None 

164 

165 @property 

166 def is_legacy_recurring_price(self) -> bool: 

167 return any(price.is_recurring for price in self.prices) 

168 

169 @property 

170 def has_seat_based_price(self) -> bool: 

171 return any( 

172 price.amount_type == ProductPriceAmountType.seat_based 

173 for price in self.prices 

174 ) 

175 

176 @hybrid_property 

177 def is_recurring(self) -> bool: 

178 if self.recurring_interval is not None: 

179 return True 

180 

181 # Check for Products that have legacy prices where recurring interval was set on them 

182 return all(price.is_recurring for price in self.prices) 

183 

184 @is_recurring.inplace.expression 

185 @classmethod 

186 def _is_recurring_expression(cls) -> ColumnElement[bool]: 

187 return or_( 

188 cls.recurring_interval.is_not(None), 

189 # Check for Products that have legacy prices where recurring interval was set on them 

190 cls.id.in_( 

191 select(ProductPrice.product_id).where( 

192 ProductPrice.type == ProductPriceType.recurring, 

193 ProductPrice.is_archived.is_(False), 

194 ) 

195 ), 

196 ) 

197 

198 @hybrid_property 

199 def billing_type(self) -> ProductBillingType: 

200 if self.is_recurring: 

201 return ProductBillingType.recurring 

202 return ProductBillingType.one_time 

203 

204 @billing_type.inplace.expression 

205 @classmethod 

206 def _billing_type_expression(cls) -> ColumnElement[ProductBillingType]: 

207 return case( 

208 (cls.is_recurring.is_(True), ProductBillingType.recurring), 

209 else_=ProductBillingType.one_time, 

210 )