Coverage for polar/models/order.py: 68%

1import inspect 

2from datetime import datetime 

3from enum import StrEnum 

4from typing import TYPE_CHECKING 

5from uuid import UUID 

6 

7from sqlalchemy import ( 

8 TIMESTAMP, 

9 ColumnElement, 

10 ForeignKey, 

11 Index, 

12 Integer, 

13 String, 

14 Uuid, 

15 func, 

16 text, 

17) 

18from sqlalchemy.dialects.postgresql import JSONB 

19from sqlalchemy.ext.associationproxy import AssociationProxy, association_proxy 

20from sqlalchemy.ext.hybrid import hybrid_property 

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

22 

23from polar.custom_field.data import CustomFieldDataMixin 

24from polar.exceptions import PolarError 

25from polar.kit.address import Address, AddressType 

26from polar.kit.db.models import RecordModel 

27from polar.kit.metadata import MetadataMixin 

28from polar.kit.tax import TaxabilityReason, TaxID, TaxIDType, TaxRate 

29from polar.models.order_item import OrderItem 

30 

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

32 from polar.models import ( 

33 Checkout, 

34 Customer, 

35 CustomerSeat, 

36 Discount, 

37 Organization, 

38 Product, 

39 ProductPrice, 

40 Subscription, 

41 ) 

42 

43 

44class OrderBillingReasonInternal(StrEnum): 

45 """ 

46 Internal billing reasons with additional granularity. 

47 """ 

48 

49 purchase = "purchase" 

50 subscription_create = "subscription_create" 

51 subscription_cycle = "subscription_cycle" 

52 subscription_cycle_after_trial = "subscription_cycle_after_trial" 

53 subscription_update = "subscription_update" 

54 

55 

56class OrderBillingReason(StrEnum): 

57 purchase = "purchase" 

58 subscription_create = "subscription_create" 

59 subscription_cycle = "subscription_cycle" 

60 subscription_update = "subscription_update" 

61 

62 

63class OrderStatus(StrEnum): 

64 pending = "pending" 

65 paid = "paid" 

66 refunded = "refunded" 

67 partially_refunded = "partially_refunded" 

68 

69 

70class OrderRefundExceedsBalance(PolarError): 

71 def __init__( 

72 self, order: "Order", attempted_amount: int, attempted_tax_amount: int 

73 ) -> None: 

74 self.type = type 

75 super().__init__( 

76 inspect.cleandoc( 

77 f""" 

78 Order({order.id}) with remaining amount {order.refundable_amount} 

79 and tax {order.refundable_tax_amount} attempted to be refunded with 

80 {attempted_amount} and {attempted_tax_amount} in taxes. 

81 """ 

82 ) 

83 ) 

84 

85 

86class Order(CustomFieldDataMixin, MetadataMixin, RecordModel): 

87 __tablename__ = "orders" 

88 __table_args__ = ( 

89 Index("ix_net_amount", text("(subtotal_amount - discount_amount)")), 

90 Index( 

91 "ix_total_amount", text("(subtotal_amount - discount_amount + tax_amount)") 

92 ), 

93 ) 

94 

95 status: Mapped[OrderStatus] = mapped_column( 

96 String, nullable=False, default=OrderStatus.pending, index=True 

97 ) 

98 subtotal_amount: Mapped[int] = mapped_column(Integer, nullable=False) 

99 discount_amount: Mapped[int] = mapped_column(Integer, nullable=False, default=0) 

100 tax_amount: Mapped[int] = mapped_column(Integer, nullable=False) 

101 applied_balance_amount: Mapped[int] = mapped_column( 

102 Integer, nullable=False, default=0 

103 ) 

104 currency: Mapped[str] = mapped_column(String(3), nullable=False) 

105 billing_reason: Mapped[OrderBillingReasonInternal] = mapped_column( 

106 String, nullable=False, index=True 

107 ) 

108 

109 refunded_amount: Mapped[int] = mapped_column(Integer, nullable=False, default=0) 

110 refunded_tax_amount: Mapped[int] = mapped_column(Integer, nullable=False, default=0) 

111 

112 platform_fee_amount: Mapped[int] = mapped_column(Integer, nullable=False, default=0) 

113 

114 billing_name: Mapped[str | None] = mapped_column( 

115 String, nullable=True, default=None 

116 ) 

117 billing_address: Mapped[Address | None] = mapped_column(AddressType, nullable=True) 

118 stripe_invoice_id: Mapped[str | None] = mapped_column( 

119 String, nullable=True, unique=True, default=None 

120 ) 

121 

122 taxability_reason: Mapped[TaxabilityReason | None] = mapped_column( 

123 String, nullable=True, default=None 

124 ) 

125 tax_id: Mapped[TaxID | None] = mapped_column(TaxIDType, nullable=True, default=None) 

126 tax_rate: Mapped[TaxRate | None] = mapped_column( 

127 JSONB(none_as_null=True), nullable=True, default=None 

128 ) 

129 tax_calculation_processor_id: Mapped[str | None] = mapped_column( 

130 String, nullable=True, default=None 

131 ) 

132 tax_transaction_processor_id: Mapped[str | None] = mapped_column( 

133 String, nullable=True, default=None 

134 ) 

135 

136 invoice_number: Mapped[str] = mapped_column(String, nullable=False, unique=True) 

137 invoice_path: Mapped[str | None] = mapped_column( 

138 String, nullable=True, default=None 

139 ) 

140 

141 next_payment_attempt_at: Mapped[datetime | None] = mapped_column( 

142 TIMESTAMP(timezone=True), nullable=True, default=None, index=True 

143 ) 

144 

145 payment_lock_acquired_at: Mapped[datetime | None] = mapped_column( 

146 TIMESTAMP(timezone=True), nullable=True, default=None 

147 ) 

148 

149 customer_id: Mapped[UUID] = mapped_column( 

150 Uuid, ForeignKey("customers.id"), nullable=False, index=True 

151 ) 

152 

153 @declared_attr 

154 def customer(cls) -> Mapped["Customer"]: 

155 return relationship("Customer", lazy="raise") 

156 

157 product_id: Mapped[UUID | None] = mapped_column( 

158 Uuid, ForeignKey("products.id"), nullable=True, index=True 

159 ) 

160 

161 @declared_attr 

162 def product(cls) -> Mapped["Product | None"]: 

163 return relationship("Product", lazy="raise") 

164 

165 discount_id: Mapped[UUID | None] = mapped_column( 

166 Uuid, ForeignKey("discounts.id", ondelete="set null"), nullable=True 

167 ) 

168 

169 @declared_attr 

170 def discount(cls) -> Mapped["Discount | None"]: 

171 return relationship("Discount", lazy="raise") 

172 

173 organization: AssociationProxy["Organization"] = association_proxy( 

174 "customer", "organization" 

175 ) 

176 

177 subscription_id: Mapped[UUID | None] = mapped_column( 

178 Uuid, ForeignKey("subscriptions.id"), nullable=True 

179 ) 

180 

181 @declared_attr 

182 def subscription(cls) -> Mapped["Subscription | None"]: 

183 return relationship("Subscription", lazy="raise") 

184 

185 checkout_id: Mapped[UUID | None] = mapped_column( 

186 Uuid, ForeignKey("checkouts.id", ondelete="set null"), nullable=True, index=True 

187 ) 

188 

189 @declared_attr 

190 def checkout(cls) -> Mapped["Checkout | None"]: 

191 return relationship("Checkout", lazy="raise") 

192 

193 seats: Mapped[int | None] = mapped_column(Integer, nullable=True, default=None) 

194 

195 items: Mapped[list["OrderItem"]] = relationship( 

196 "OrderItem", 

197 back_populates="order", 

198 cascade="all, delete-orphan", 

199 # Items are almost always needed, so eager loading makes sense 

200 lazy="selectin", 

201 ) 

202 

203 @declared_attr 

204 def customer_seats(cls) -> Mapped[list["CustomerSeat"]]: 

205 return relationship( 

206 "CustomerSeat", 

207 lazy="raise", 

208 back_populates="order", 

209 cascade="all, delete-orphan", 

210 ) 

211 

212 @property 

213 def legacy_product_price(self) -> "ProductPrice | None": 

214 """ 

215 Dummy method to keep API backward compatibility 

216 by fetching a product price at all costs. 

217 """ 

218 if self.product is None: 

219 return None 

220 for item in self.items: 

221 if item.product_price: 

222 return item.product_price 

223 return self.product.prices[0] 

224 

225 @property 

226 def legacy_product_price_id(self) -> UUID | None: 

227 price = self.legacy_product_price 

228 if price is None: 

229 return None 

230 return price.id 

231 

232 @hybrid_property 

233 def paid(self) -> bool: 

234 return self.status in { 

235 OrderStatus.paid, 

236 OrderStatus.refunded, 

237 OrderStatus.partially_refunded, 

238 } 

239 

240 @paid.inplace.expression 

241 @classmethod 

242 def _paid_expression(cls) -> ColumnElement[bool]: 

243 return cls.status.in_( 

244 (OrderStatus.paid, OrderStatus.refunded, OrderStatus.partially_refunded) 

245 ) 

246 

247 @hybrid_property 

248 def net_amount(self) -> int: 

249 return self.subtotal_amount - self.discount_amount 

250 

251 @net_amount.inplace.expression 

252 @classmethod 

253 def _net_amount_expression(cls) -> ColumnElement[int]: 

254 return cls.subtotal_amount - cls.discount_amount 

255 

256 @hybrid_property 

257 def total_amount(self) -> int: 

258 return self.net_amount + self.tax_amount 

259 

260 @total_amount.inplace.expression 

261 @classmethod 

262 def _total_amount_expression(cls) -> ColumnElement[int]: 

263 return cls.net_amount + cls.tax_amount 

264 

265 @hybrid_property 

266 def due_amount(self) -> int: 

267 return max(0, self.total_amount + self.applied_balance_amount) 

268 

269 @due_amount.inplace.expression 

270 @classmethod 

271 def _due_amount_expression(cls) -> ColumnElement[int]: 

272 return func.greatest(0, cls.total_amount + cls.applied_balance_amount) 

273 

274 @hybrid_property 

275 def payout_amount(self) -> int: 

276 return self.net_amount - self.platform_fee_amount - self.refunded_amount 

277 

278 @payout_amount.inplace.expression 

279 @classmethod 

280 def _payout_amount_expression(cls) -> ColumnElement[int]: 

281 return cls.net_amount - cls.platform_fee_amount - cls.refunded_amount 

282 

283 @property 

284 def taxed(self) -> int: 

285 return self.tax_amount > 0 

286 

287 @property 

288 def refunded(self) -> bool: 

289 return self.status == OrderStatus.refunded 

290 

291 @property 

292 def refundable_amount(self) -> int: 

293 return self.net_amount - self.refunded_amount 

294 

295 @property 

296 def refundable_tax_amount(self) -> int: 

297 return self.tax_amount - self.refunded_tax_amount 

298 

299 @property 

300 def remaining_balance(self) -> int: 

301 return self.refundable_amount + self.refundable_tax_amount 

302 

303 def update_refunds(self, refunded_amount: int, refunded_tax_amount: int) -> None: 

304 new_amount = self.refunded_amount + refunded_amount 

305 new_tax_amount = self.refunded_tax_amount + refunded_tax_amount 

306 exceeds_original_amount = ( 

307 new_amount < 0 

308 or new_amount > self.net_amount 

309 or new_tax_amount < 0 

310 or new_tax_amount > self.tax_amount 

311 ) 

312 if exceeds_original_amount: 

313 raise OrderRefundExceedsBalance(self, refunded_amount, refunded_tax_amount) 

314 

315 if new_amount == 0: 

316 new_status = OrderStatus.paid 

317 elif new_amount == self.net_amount: 

318 new_status = OrderStatus.refunded 

319 else: 

320 new_status = OrderStatus.partially_refunded 

321 

322 self.status = new_status 

323 self.refunded_amount = new_amount 

324 self.refunded_tax_amount = new_tax_amount 

325 

326 @property 

327 def is_invoice_generated(self) -> bool: 

328 return self.invoice_path is not None 

329 

330 @property 

331 def invoice_filename(self) -> str: 

332 return f"Invoice-{self.invoice_number}.pdf" 

333 

334 @property 

335 def statement_descriptor_suffix(self) -> str: 

336 if ( 

337 self.billing_reason 

338 == OrderBillingReasonInternal.subscription_cycle_after_trial 

339 ): 

340 return self.organization.statement_descriptor(" TRIAL OVER") 

341 return self.organization.statement_descriptor() 

342 

343 @property 

344 def description(self) -> str: 

345 if self.product is not None: 

346 return self.product.name 

347 return self.items[0].label