Coverage for polar/subscription/schemas.py: 95%
102 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 16:17 +0000
1import inspect 1a
2from datetime import datetime 1a
3from typing import Annotated, Literal 1a
5from babel.numbers import format_currency 1a
6from fastapi import Path 1a
7from pydantic import UUID4, AliasChoices, AliasPath, Field, FutureDatetime 1a
8from pydantic.json_schema import SkipJsonSchema 1a
10from polar.custom_field.data import CustomFieldDataOutputMixin 1a
11from polar.customer.schemas.customer import CustomerBase 1a
12from polar.discount.schemas import DiscountMinimal 1a
13from polar.enums import SubscriptionProrationBehavior, SubscriptionRecurringInterval 1a
14from polar.kit.metadata import MetadataInputMixin, MetadataOutputMixin 1a
15from polar.kit.schemas import ( 1a
16 CUSTOMER_ID_EXAMPLE,
17 METER_ID_EXAMPLE,
18 PRODUCT_ID_EXAMPLE,
19 IDSchema,
20 MergeJSONSchema,
21 Schema,
22 SetSchemaReference,
23 TimestampedSchema,
24)
25from polar.meter.schemas import Meter 1a
26from polar.models.subscription import CustomerCancellationReason, SubscriptionStatus 1a
27from polar.product.schemas import Product, ProductPrice 1a
29SubscriptionID = Annotated[UUID4, Path(description="The subscription ID.")] 1a
32class SubscriptionCustomer(CustomerBase): ... 1a
35class SubscriptionUser(Schema): 1a
36 id: UUID4 = Field( 1a
37 validation_alias=AliasChoices(
38 # Validate from ORM model
39 "legacy_user_id",
40 # Validate from stored webhook payload
41 "id",
42 )
43 )
44 email: str 1a
45 public_name: str = Field( 1a
46 validation_alias=AliasChoices(
47 # Validate from ORM model
48 "legacy_user_public_name",
49 # Validate from stored webhook payload
50 "public_name",
51 )
52 )
53 avatar_url: str | None = Field(None) 1a
54 github_username: str | None = Field(None) 1a
57class SubscriptionBase(IDSchema, TimestampedSchema): 1a
58 amount: int = Field(description="The amount of the subscription.", examples=[10000]) 1a
59 currency: str = Field( 1a
60 description="The currency of the subscription.", examples=["usd"]
61 )
62 recurring_interval: SubscriptionRecurringInterval = Field( 1a
63 description="The interval at which the subscription recurs.",
64 examples=["month"],
65 )
66 recurring_interval_count: int = Field( 1a
67 description=(
68 "Number of interval units of the subscription. "
69 "If this is set to 1 the charge will happen every interval (e.g. every month), "
70 "if set to 2 it will be every other month, and so on."
71 )
72 )
73 status: SubscriptionStatus = Field( 1a
74 description="The status of the subscription.", examples=["active"]
75 )
76 current_period_start: datetime = Field( 1a
77 description="The start timestamp of the current billing period."
78 )
79 current_period_end: datetime | None = Field( 1a
80 description="The end timestamp of the current billing period."
81 )
82 trial_start: datetime | None = Field( 1a
83 description="The start timestamp of the trial period, if any."
84 )
85 trial_end: datetime | None = Field( 1a
86 description="The end timestamp of the trial period, if any."
87 )
88 cancel_at_period_end: bool = Field( 1a
89 description=(
90 "Whether the subscription will be canceled "
91 "at the end of the current period."
92 )
93 )
94 canceled_at: datetime | None = Field( 1a
95 description=(
96 "The timestamp when the subscription was canceled. "
97 "The subscription might still be active if `cancel_at_period_end` is `true`."
98 )
99 )
100 started_at: datetime | None = Field( 1a
101 description="The timestamp when the subscription started."
102 )
103 ends_at: datetime | None = Field( 1a
104 description="The timestamp when the subscription will end."
105 )
106 ended_at: datetime | None = Field( 1a
107 description="The timestamp when the subscription ended."
108 )
110 customer_id: UUID4 = Field(description="The ID of the subscribed customer.") 1a
111 product_id: UUID4 = Field(description="The ID of the subscribed product.") 1a
112 discount_id: UUID4 | None = Field( 1a
113 description="The ID of the applied discount, if any."
114 )
115 checkout_id: UUID4 | None 1a
117 seats: int | None = Field( 1a
118 default=None,
119 description="The number of seats for seat-based subscriptions. None for non-seat subscriptions.",
120 )
122 customer_cancellation_reason: CustomerCancellationReason | None 1a
123 customer_cancellation_comment: str | None 1a
125 price_id: SkipJsonSchema[UUID4] = Field( 1a
126 deprecated="Use `prices` instead.",
127 validation_alias=AliasChoices(
128 # Validate from stored webhook payload
129 "price_id",
130 # Validate from ORM model
131 AliasPath("prices", 0, "id"),
132 ),
133 )
135 def get_amount_display(self) -> str: 1a
136 if self.amount is None or self.currency is None:
137 return "Free"
138 return f"{
139 format_currency(
140 self.amount / 100,
141 self.currency.upper(),
142 locale='en_US',
143 )
144 }/{self.recurring_interval}"
147SubscriptionDiscount = Annotated[ 1a
148 DiscountMinimal, MergeJSONSchema({"title": "SubscriptionDiscount"})
149]
152class SubscriptionMeterBase(IDSchema, TimestampedSchema): 1a
153 consumed_units: float = Field( 1a
154 description="The number of consumed units so far in this billing period.",
155 examples=[25.0],
156 )
157 credited_units: int = Field( 1a
158 description="The number of credited units so far in this billing period.",
159 examples=[100],
160 )
161 amount: int = Field( 1a
162 description="The amount due in cents so far in this billing period.",
163 examples=[0],
164 )
165 meter_id: UUID4 = Field( 1a
166 description="The ID of the meter.", examples=[METER_ID_EXAMPLE]
167 )
170class SubscriptionMeter(SubscriptionMeterBase): 1a
171 """Current consumption and spending for a subscription meter."""
173 meter: Meter = Field( 1a
174 description="The meter associated with this subscription.",
175 )
178class Subscription(CustomFieldDataOutputMixin, MetadataOutputMixin, SubscriptionBase): 1a
179 customer: SubscriptionCustomer 1a
180 user_id: SkipJsonSchema[UUID4] = Field( 1a
181 validation_alias=AliasChoices(
182 # Validate from stored webhook payload
183 "user_id",
184 # Validate from ORM model
185 AliasPath("customer", "legacy_user_id"),
186 ),
187 deprecated="Use `customer_id`.",
188 )
189 user: SkipJsonSchema[SubscriptionUser] = Field( 1a
190 validation_alias=AliasChoices(
191 # Validate from stored webhook payload
192 "user",
193 # Validate from ORM model
194 "customer",
195 ),
196 deprecated="Use `customer`.",
197 )
198 product: Product 1a
199 discount: SubscriptionDiscount | None 1a
201 price: SkipJsonSchema[ProductPrice] = Field( 1a
202 deprecated="Use `prices` instead.",
203 validation_alias=AliasChoices(
204 # Validate from stored webhook payload
205 "price",
206 # Validate from ORM model
207 AliasPath("prices", 0),
208 ),
209 )
211 prices: list[ProductPrice] = Field( 1a
212 description="List of enabled prices for the subscription."
213 )
214 meters: list[SubscriptionMeter] = Field( 1a
215 description="List of meters associated with the subscription."
216 )
219class SubscriptionCreateBase(MetadataInputMixin, Schema): 1a
220 product_id: UUID4 = Field( 1a
221 description=(
222 "The ID of the recurring product to subscribe to. "
223 "Must be a free product, otherwise the customer should go through a checkout flow."
224 ),
225 examples=[PRODUCT_ID_EXAMPLE],
226 )
229class SubscriptionCreateCustomer(SubscriptionCreateBase): 1a
230 """
231 Create a subscription for an existing customer.
232 """
234 customer_id: UUID4 = Field( 1a
235 description="The ID of the customer to create the subscription for.",
236 examples=[CUSTOMER_ID_EXAMPLE],
237 )
240class SubscriptionCreateExternalCustomer(SubscriptionCreateBase): 1a
241 """
242 Create a subscription for an existing customer identified by an external ID.
243 """
245 external_customer_id: str = Field( 1a
246 description=(
247 "The ID of the customer in your system to create the subscription for. "
248 "It must already exist in Polar."
249 )
250 )
253SubscriptionCreate = SubscriptionCreateCustomer | SubscriptionCreateExternalCustomer 1a
256class SubscriptionUpdateProduct(Schema): 1a
257 product_id: UUID4 = Field( 1a
258 description="Update subscription to another product.",
259 examples=[PRODUCT_ID_EXAMPLE],
260 )
261 proration_behavior: SubscriptionProrationBehavior | None = Field( 1a
262 default=None,
263 description=(
264 "Determine how to handle the proration billing. "
265 "If not provided, will use the default organization setting."
266 ),
267 )
270class SubscriptionUpdateDiscount(Schema): 1a
271 discount_id: UUID4 | None = Field( 1a
272 description=(
273 "Update the subscription to apply a new discount. "
274 "If set to `null`, the discount will be removed."
275 " The change will be applied on the next billing cycle."
276 ),
277 )
280class SubscriptionUpdateTrial(Schema): 1a
281 trial_end: FutureDatetime | Literal["now"] = Field( 1a
282 description=(
283 "Set or extend the trial period of the subscription. "
284 "If set to `now`, the trial will end immediately."
285 ),
286 )
289class SubscriptionUpdateSeats(Schema): 1a
290 seats: int = Field( 1a
291 description="Update the number of seats for this subscription.",
292 ge=1,
293 )
294 proration_behavior: SubscriptionProrationBehavior | None = Field( 1a
295 default=None,
296 description=(
297 "Determine how to handle the proration billing. "
298 "If not provided, will use the default organization setting."
299 ),
300 )
303class SubscriptionUpdateBillingPeriod(Schema): 1a
304 current_billing_period_end: FutureDatetime = Field( 1a
305 description=inspect.cleandoc(
306 """
307 Set a new date for the end of the current billing period. The subscription will renew on this date. Needs to be later than the current value.
309 It is not possible to update the current billing period on a canceled subscription.
310 """
311 )
312 )
315class SubscriptionCancelBase(Schema): 1a
316 customer_cancellation_reason: CustomerCancellationReason | None = Field( 1a
317 None,
318 description=inspect.cleandoc(
319 """
320 Customer reason for cancellation.
322 Helpful to monitor reasons behind churn for future improvements.
324 Only set this in case your own service is requesting the reason from the
325 customer. Or you know based on direct conversations, i.e support, with
326 the customer.
328 * `too_expensive`: Too expensive for the customer.
329 * `missing_features`: Customer is missing certain features.
330 * `switched_service`: Customer switched to another service.
331 * `unused`: Customer is not using it enough.
332 * `customer_service`: Customer is not satisfied with the customer service.
333 * `low_quality`: Customer is unhappy with the quality.
334 * `too_complex`: Customer considers the service too complicated.
335 * `other`: Other reason(s).
336 """
337 ),
338 )
339 customer_cancellation_comment: str | None = Field( 1a
340 None,
341 description=inspect.cleandoc(
342 """
343 Customer feedback and why they decided to cancel.
345 **IMPORTANT:**
346 Do not use this to store internal notes! It's intended to be input
347 from the customer and is therefore also available in their Polar
348 purchases library.
350 Only set this in case your own service is requesting the reason from the
351 customer. Or you copy a message directly from a customer
352 conversation, i.e support.
353 """
354 ),
355 )
358class SubscriptionCancel(SubscriptionCancelBase): 1a
359 cancel_at_period_end: bool = Field( 1a
360 description=inspect.cleandoc(
361 """
362 Cancel an active subscription once the current period ends.
364 Or uncancel a subscription currently set to be revoked at period end.
365 """
366 ),
367 )
370class SubscriptionRevoke(SubscriptionCancelBase): 1a
371 revoke: Literal[True] = Field( 1a
372 description="Cancel and revoke an active subscription immediately"
373 )
376SubscriptionUpdate = Annotated[ 1a
377 SubscriptionUpdateProduct
378 | SubscriptionUpdateDiscount
379 | SubscriptionUpdateTrial
380 | SubscriptionUpdateSeats
381 | SubscriptionUpdateBillingPeriod
382 | SubscriptionCancel
383 | SubscriptionRevoke,
384 SetSchemaReference("SubscriptionUpdate"),
385]
388class SubscriptionChargePreview(Schema): 1a
389 """Preview of the next charge for a subscription."""
391 base_amount: int = Field( 1a
392 description="Base subscription amount in cents (sum of product prices)"
393 )
394 metered_amount: int = Field( 1a
395 description="Total metered usage charges in cents (sum of all meter charges)"
396 )
397 subtotal_amount: int = Field( 1a
398 description="Subtotal amount in cents (base + metered, before discount and tax)"
399 )
400 discount_amount: int = Field(description="Discount amount in cents") 1a
401 tax_amount: int = Field(description="Tax amount in cents") 1a
402 total_amount: int = Field(description="Total amount in cents (final charge amount)") 1a