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

1import inspect 2 ctx1ab

2from datetime import datetime 2 ctx1ab

3from enum import StrEnum 2 ctx1ab

4from typing import TYPE_CHECKING 2 ctx1ab

5from uuid import UUID 2 ctx1ab

6 

7from sqlalchemy import ( 2 ctx1ab

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 2 ctx1ab

19from sqlalchemy.ext.associationproxy import AssociationProxy, association_proxy 2 ctx1ab

20from sqlalchemy.ext.hybrid import hybrid_property 2 ctx1ab

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

22 

23from polar.custom_field.data import CustomFieldDataMixin 2 ctx1ab

24from polar.exceptions import PolarError 2 ctx1ab

25from polar.kit.address import Address, AddressType 2 ctx1ab

26from polar.kit.db.models import RecordModel 2 ctx1ab

27from polar.kit.metadata import MetadataMixin 2 ctx1ab

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

29from polar.models.order_item import OrderItem 2 ctx1ab

30 

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

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): 2 ctx1ab

45 """ 

46 Internal billing reasons with additional granularity. 

47 """ 

48 

49 purchase = "purchase" 2 ctx1ab

50 subscription_create = "subscription_create" 2 ctx1ab

51 subscription_cycle = "subscription_cycle" 2 ctx1ab

52 subscription_cycle_after_trial = "subscription_cycle_after_trial" 2 ctx1ab

53 subscription_update = "subscription_update" 2 ctx1ab

54 

55 

56class OrderBillingReason(StrEnum): 2 ctx1ab

57 purchase = "purchase" 2 ctx1ab

58 subscription_create = "subscription_create" 2 ctx1ab

59 subscription_cycle = "subscription_cycle" 2 ctx1ab

60 subscription_update = "subscription_update" 2 ctx1ab

61 

62 

63class OrderStatus(StrEnum): 2 ctx1ab

64 pending = "pending" 2 ctx1ab

65 paid = "paid" 2 ctx1ab

66 refunded = "refunded" 2 ctx1ab

67 partially_refunded = "partially_refunded" 2 ctx1ab

68 

69 

70class OrderRefundExceedsBalance(PolarError): 2 ctx1ab

71 def __init__( 2 ctx1ab

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): 2 ctx1ab

87 __tablename__ = "orders" 2 ctx1ab

88 __table_args__ = ( 2 ctx1ab

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( 2 ctx1ab

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

97 ) 

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

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

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

101 applied_balance_amount: Mapped[int] = mapped_column( 2 ctx1ab

102 Integer, nullable=False, default=0 

103 ) 

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

105 billing_reason: Mapped[OrderBillingReasonInternal] = mapped_column( 2 ctx1ab

106 String, nullable=False, index=True 

107 ) 

108 

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

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

111 

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

113 

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

115 String, nullable=True, default=None 

116 ) 

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

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

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

120 ) 

121 

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

123 String, nullable=True, default=None 

124 ) 

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

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

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

128 ) 

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

130 String, nullable=True, default=None 

131 ) 

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

133 String, nullable=True, default=None 

134 ) 

135 

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

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

138 String, nullable=True, default=None 

139 ) 

140 

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

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

143 ) 

144 

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

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

147 ) 

148 

149 customer_id: Mapped[UUID] = mapped_column( 2 ctx1ab

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

151 ) 

152 

153 @declared_attr 2 ctx1ab

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

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

156 

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

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

159 ) 

160 

161 @declared_attr 2 ctx1ab

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

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

164 

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

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

167 ) 

168 

169 @declared_attr 2 ctx1ab

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

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

172 

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

174 "customer", "organization" 

175 ) 

176 

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

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

179 ) 

180 

181 @declared_attr 2 ctx1ab

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

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

184 

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

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

187 ) 

188 

189 @declared_attr 2 ctx1ab

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

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

192 

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

194 

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

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 2 ctx1ab

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

205 return relationship( 2 ctx1ab

206 "CustomerSeat", 

207 lazy="raise", 

208 back_populates="order", 

209 cascade="all, delete-orphan", 

210 ) 

211 

212 @property 2 ctx1ab

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

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 2 ctx1ab

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

227 price = self.legacy_product_price 

228 if price is None: 

229 return None 

230 return price.id 

231 

232 @hybrid_property 2 ctx1ab

233 def paid(self) -> bool: 2 ctx1ab

234 return self.status in { 

235 OrderStatus.paid, 

236 OrderStatus.refunded, 

237 OrderStatus.partially_refunded, 

238 } 

239 

240 @paid.inplace.expression 2 ctx1ab

241 @classmethod 2 ctx1ab

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

243 return cls.status.in_( 

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

245 ) 

246 

247 @hybrid_property 2 ctx1ab

248 def net_amount(self) -> int: 2 ctx1ab

249 return self.subtotal_amount - self.discount_amount 

250 

251 @net_amount.inplace.expression 2 ctx1ab

252 @classmethod 2 ctx1ab

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

254 return cls.subtotal_amount - cls.discount_amount 

255 

256 @hybrid_property 2 ctx1ab

257 def total_amount(self) -> int: 2 ctx1ab

258 return self.net_amount + self.tax_amount 

259 

260 @total_amount.inplace.expression 2 ctx1ab

261 @classmethod 2 ctx1ab

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

263 return cls.net_amount + cls.tax_amount 

264 

265 @hybrid_property 2 ctx1ab

266 def due_amount(self) -> int: 2 ctx1ab

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

268 

269 @due_amount.inplace.expression 2 ctx1ab

270 @classmethod 2 ctx1ab

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

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

273 

274 @hybrid_property 2 ctx1ab

275 def payout_amount(self) -> int: 2 ctx1ab

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

277 

278 @payout_amount.inplace.expression 2 ctx1ab

279 @classmethod 2 ctx1ab

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

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

282 

283 @property 2 ctx1ab

284 def taxed(self) -> int: 2 ctx1ab

285 return self.tax_amount > 0 

286 

287 @property 2 ctx1ab

288 def refunded(self) -> bool: 2 ctx1ab

289 return self.status == OrderStatus.refunded 

290 

291 @property 2 ctx1ab

292 def refundable_amount(self) -> int: 2 ctx1ab

293 return self.net_amount - self.refunded_amount 

294 

295 @property 2 ctx1ab

296 def refundable_tax_amount(self) -> int: 2 ctx1ab

297 return self.tax_amount - self.refunded_tax_amount 

298 

299 @property 2 ctx1ab

300 def remaining_balance(self) -> int: 2 ctx1ab

301 return self.refundable_amount + self.refundable_tax_amount 

302 

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

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 2 ctx1ab

327 def is_invoice_generated(self) -> bool: 2 ctx1ab

328 return self.invoice_path is not None 

329 

330 @property 2 ctx1ab

331 def invoice_filename(self) -> str: 2 ctx1ab

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

333 

334 @property 2 ctx1ab

335 def statement_descriptor_suffix(self) -> str: 2 ctx1ab

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 2 ctx1ab

344 def description(self) -> str: 2 ctx1ab

345 if self.product is not None: 

346 return self.product.name 

347 return self.items[0].label