Coverage for polar/order/schemas.py: 91%

91 statements  

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

1from typing import Annotated 1a

2 

3from babel.numbers import format_currency 1a

4from fastapi import Path 1a

5from pydantic import ( 1a

6 UUID4, 

7 AliasChoices, 

8 AliasPath, 

9 Field, 

10 computed_field, 

11 field_serializer, 

12) 

13from pydantic.json_schema import SkipJsonSchema 1a

14 

15from polar.custom_field.data import CustomFieldDataOutputMixin 1a

16from polar.customer.schemas.customer import CustomerBase 1a

17from polar.discount.schemas import DiscountMinimal 1a

18from polar.exceptions import ResourceNotFound 1a

19from polar.kit.address import Address, AddressInput 1a

20from polar.kit.metadata import MetadataOutputMixin 1a

21from polar.kit.schemas import IDSchema, MergeJSONSchema, Schema, TimestampedSchema 1a

22from polar.models.order import ( 1a

23 OrderBillingReason, 

24 OrderBillingReasonInternal, 

25 OrderStatus, 

26) 

27from polar.product.schemas import ProductBase, ProductPrice 1a

28from polar.subscription.schemas import SubscriptionBase 1a

29 

30OrderID = Annotated[UUID4, Path(description="The order ID.")] 1a

31 

32OrderNotFound = { 1a

33 "description": "Order not found.", 

34 "model": ResourceNotFound.schema(), 

35} 

36 

37 

38class OrderBase(TimestampedSchema, IDSchema): 1a

39 status: OrderStatus = Field(examples=["paid"]) 1a

40 paid: bool = Field( 1a

41 description="Whether the order has been paid for.", examples=[True] 

42 ) 

43 subtotal_amount: int = Field( 1a

44 description="Amount in cents, before discounts and taxes.", examples=[10000] 

45 ) 

46 discount_amount: int = Field( 1a

47 description="Discount amount in cents.", examples=[1000] 

48 ) 

49 net_amount: int = Field( 1a

50 description="Amount in cents, after discounts but before taxes.", 

51 examples=[9000], 

52 ) 

53 

54 tax_amount: int = Field(description="Sales tax amount in cents.", examples=[720]) 1a

55 total_amount: int = Field( 1a

56 description="Amount in cents, after discounts and taxes.", examples=[9720] 

57 ) 

58 

59 applied_balance_amount: int = Field( 1a

60 description=( 

61 "Customer's balance amount applied to this invoice. " 

62 "Can increase the total amount paid, if the customer has a negative balance, " 

63 " or decrease it, if the customer has a positive balance." 

64 "Amount in cents." 

65 ), 

66 examples=[0], 

67 ) 

68 due_amount: int = Field( 1a

69 description="Amount in cents that is due for this order.", examples=[0] 

70 ) 

71 refunded_amount: int = Field(description="Amount refunded in cents.", examples=[0]) 1a

72 refunded_tax_amount: int = Field( 1a

73 description="Sales tax refunded in cents.", examples=[0] 

74 ) 

75 currency: str = Field(examples=["usd"]) 1a

76 billing_reason: OrderBillingReasonInternal 1a

77 billing_name: str | None = Field( 1a

78 description="The name of the customer that should appear on the invoice. " 

79 ) 

80 billing_address: Address | None 1a

81 

82 @field_serializer("billing_reason") 1a

83 def serialize_billing_reason( 1a

84 self, value: OrderBillingReasonInternal 

85 ) -> OrderBillingReason: 

86 if value == OrderBillingReasonInternal.subscription_cycle_after_trial: 

87 return OrderBillingReason.subscription_cycle 

88 return OrderBillingReason(value) 

89 

90 invoice_number: str = Field( 1a

91 description="The invoice number associated with this order." 

92 ) 

93 is_invoice_generated: bool = Field( 1a

94 description="Whether an invoice has been generated for this order." 

95 ) 

96 

97 seats: int | None = Field( 1a

98 None, description="Number of seats purchased (for seat-based one-time orders)." 

99 ) 

100 

101 customer_id: UUID4 1a

102 product_id: UUID4 | None 1a

103 product_price_id: SkipJsonSchema[UUID4 | None] = Field( 1a

104 deprecated="Use `items` instead.", 

105 validation_alias=AliasChoices( 

106 # Validate from stored webhook payload 

107 "product_price_id", 

108 # Validate from ORM model 

109 "legacy_product_price_id", 

110 ), 

111 ) 

112 discount_id: UUID4 | None 1a

113 subscription_id: UUID4 | None 1a

114 checkout_id: UUID4 | None 1a

115 

116 @computed_field( 1a

117 description="Amount in cents, after discounts but before taxes.", 

118 deprecated=( 

119 "Use `net_amount`. " 

120 "It has the same value and meaning, but the name is more descriptive." 

121 ), 

122 ) 

123 def amount(self) -> SkipJsonSchema[int]: 1a

124 return self.net_amount 

125 

126 @computed_field(deprecated="Use `applied_balance_amount`.") 1a

127 def from_balance_amount(self) -> SkipJsonSchema[int]: 1a

128 return self.applied_balance_amount 

129 

130 def get_amount_display(self) -> str: 1a

131 return format_currency( 

132 self.net_amount / 100, self.currency.upper(), locale="en_US" 

133 ) 

134 

135 def get_refunded_amount_display(self) -> str: 1a

136 return format_currency( 

137 self.refunded_amount / 100, self.currency.upper(), locale="en_US" 

138 ) 

139 

140 

141class OrderCustomer(CustomerBase): ... 1a

142 

143 

144class OrderUser(Schema): 1a

145 id: UUID4 = Field( 1a

146 validation_alias=AliasChoices( 

147 # Validate from ORM model 

148 "legacy_user_id", 

149 # Validate from stored webhook payload 

150 "id", 

151 ) 

152 ) 

153 email: str 1a

154 public_name: str = Field( 1a

155 validation_alias=AliasChoices( 

156 # Validate from ORM model 

157 "legacy_user_public_name", 

158 # Validate from stored webhook payload 

159 "public_name", 

160 ) 

161 ) 

162 avatar_url: str | None = Field(None) 1a

163 github_username: str | None = Field(None) 1a

164 

165 

166class OrderProduct(ProductBase, MetadataOutputMixin): ... 1a

167 

168 

169OrderDiscount = Annotated[DiscountMinimal, MergeJSONSchema({"title": "OrderDiscount"})] 1a

170 

171 

172class OrderSubscription(SubscriptionBase, MetadataOutputMixin): 1a

173 user_id: SkipJsonSchema[UUID4] = Field( 1a

174 validation_alias=AliasChoices( 

175 # Validate from stored webhook payload 

176 "user_id", 

177 # Validate from ORM model 

178 AliasPath("customer", "legacy_user_id"), 

179 ), 

180 deprecated="Use `customer_id`.", 

181 ) 

182 

183 

184class OrderItemSchema(IDSchema, TimestampedSchema): 1a

185 """ 

186 An order line item. 

187 """ 

188 

189 label: str = Field( 1a

190 description="Description of the line item charge.", examples=["Pro Plan"] 

191 ) 

192 amount: int = Field( 1a

193 description="Amount in cents, before discounts and taxes.", examples=[10000] 

194 ) 

195 tax_amount: int = Field(description="Sales tax amount in cents.", examples=[720]) 1a

196 proration: bool = Field( 1a

197 description="Whether this charge is due to a proration.", examples=[False] 

198 ) 

199 product_price_id: UUID4 | None = Field(description="Associated price ID, if any.") 1a

200 

201 

202class Order(CustomFieldDataOutputMixin, MetadataOutputMixin, OrderBase): 1a

203 platform_fee_amount: int = Field( 1a

204 description="Platform fee amount in cents.", examples=[500] 

205 ) 

206 customer: OrderCustomer 1a

207 user_id: UUID4 = Field( 1a

208 validation_alias=AliasChoices( 

209 # Validate from stored webhook payload 

210 "user_id", 

211 # Validate from ORM model 

212 AliasPath("customer", "legacy_user_id"), 

213 ), 

214 deprecated="Use `customer_id`.", 

215 ) 

216 user: SkipJsonSchema[OrderUser] = Field( 1a

217 validation_alias=AliasChoices( 

218 # Validate from stored webhook payload 

219 "user", 

220 # Validate from ORM model 

221 "customer", 

222 ), 

223 deprecated="Use `customer`.", 

224 ) 

225 product: OrderProduct | None 1a

226 product_price: SkipJsonSchema[ProductPrice | None] = Field( 1a

227 deprecated="Use `items` instead.", 

228 validation_alias=AliasChoices( 

229 # Validate from stored webhook payload 

230 "product_price", 

231 # Validate from ORM model 

232 "legacy_product_price", 

233 ), 

234 ) 

235 discount: OrderDiscount | None 1a

236 subscription: OrderSubscription | None 1a

237 items: list[OrderItemSchema] = Field(description="Line items composing the order.") 1a

238 description: str = Field( 1a

239 description="A summary description of the order.", examples=["Pro Plan"] 

240 ) 

241 

242 

243class OrderUpdateBase(Schema): 1a

244 billing_name: str | None = Field( 1a

245 description=( 

246 "The name of the customer that should appear on the invoice. " 

247 "Can't be updated after the invoice is generated." 

248 ) 

249 ) 

250 billing_address: AddressInput | None = Field( 1a

251 description=( 

252 "The address of the customer that should appear on the invoice. " 

253 "Can't be updated after the invoice is generated." 

254 ) 

255 ) 

256 

257 

258class OrderUpdate(OrderUpdateBase): 1a

259 """Schema to update an order.""" 

260 

261 

262class OrderInvoice(Schema): 1a

263 """Order's invoice data.""" 

264 

265 url: str = Field(..., description="The URL to the invoice.") 1a