Coverage for polar/customer_portal/schemas/subscription.py: 98%
40 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 15:52 +0000
1import inspect 1a
2from typing import Annotated 1a
4from pydantic import UUID4, AliasChoices, AliasPath, Field, computed_field 1a
5from pydantic.json_schema import SkipJsonSchema 1a
7from polar.enums import SubscriptionProrationBehavior 1a
8from polar.kit.schemas import IDSchema, Schema, SetSchemaReference, TimestampedSchema 1a
9from polar.meter.schemas import NAME_DESCRIPTION as METER_NAME_DESCRIPTION 1a
10from polar.models.subscription import CustomerCancellationReason 1a
11from polar.product.schemas import ( 1a
12 BenefitPublicList,
13 ProductBase,
14 ProductMediaList,
15 ProductPrice,
16 ProductPriceList,
17)
18from polar.subscription.schemas import SubscriptionBase, SubscriptionMeterBase 1a
20from .organization import CustomerOrganization 1a
23class CustomerSubscriptionProduct(ProductBase): 1a
24 prices: ProductPriceList 1a
25 benefits: BenefitPublicList 1a
26 medias: ProductMediaList 1a
27 organization: CustomerOrganization 1a
30class CustomerSubscriptionMeterMeter(IDSchema, TimestampedSchema): 1a
31 name: str = Field(description=METER_NAME_DESCRIPTION) 1a
34class CustomerSubscriptionMeter(SubscriptionMeterBase): 1a
35 meter: CustomerSubscriptionMeterMeter 1a
38class CustomerSubscription(SubscriptionBase): 1a
39 user_id: SkipJsonSchema[UUID4] = Field( 1a
40 validation_alias=AliasChoices(
41 # Validate from stored webhook payload
42 "user_id",
43 # Validate from ORM model
44 AliasPath("customer", "legacy_user_id"),
45 ),
46 deprecated="Use `customer_id`.",
47 )
48 product: CustomerSubscriptionProduct 1a
50 price: SkipJsonSchema[ProductPrice] = Field( 1a
51 deprecated="Use `prices` instead.",
52 validation_alias=AliasChoices(
53 # Validate from stored webhook payload
54 "price",
55 # Validate from ORM model
56 AliasPath("prices", 0),
57 ),
58 )
60 prices: list[ProductPrice] = Field( 1a
61 description="List of enabled prices for the subscription."
62 )
63 meters: list[CustomerSubscriptionMeter] = Field( 1a
64 description="List of meters associated with the subscription."
65 )
67 stripe_subscription_id: SkipJsonSchema[str | None] = Field( 1a
68 validation_alias="stripe_subscription_id"
69 )
71 @computed_field 1a
72 def is_polar_managed(self) -> bool: 1a
73 """Whether the subscription is managed by Polar."""
74 return self.stripe_subscription_id is None
77class CustomerSubscriptionUpdateProduct(Schema): 1a
78 product_id: UUID4 = Field(description="Update subscription to another product.") 1a
81class CustomerSubscriptionUpdateSeats(Schema): 1a
82 seats: int = Field( 1a
83 description="Update the number of seats for this subscription.",
84 ge=1,
85 )
86 proration_behavior: SubscriptionProrationBehavior | None = Field( 1a
87 default=None,
88 description=(
89 "Determine how to handle the proration billing. "
90 "If not provided, will use the default organization setting."
91 ),
92 )
95class CustomerSubscriptionCancel(Schema): 1a
96 cancel_at_period_end: bool | None = Field( 1a
97 None,
98 description=inspect.cleandoc(
99 """
100 Cancel an active subscription once the current period ends.
102 Or uncancel a subscription currently set to be revoked at period end.
103 """
104 ),
105 )
107 cancellation_reason: CustomerCancellationReason | None = Field( 1a
108 None,
109 description=inspect.cleandoc(
110 """
111 Customers reason for cancellation.
113 * `too_expensive`: Too expensive for the customer.
114 * `missing_features`: Customer is missing certain features.
115 * `switched_service`: Customer switched to another service.
116 * `unused`: Customer is not using it enough.
117 * `customer_service`: Customer is not satisfied with the customer service.
118 * `low_quality`: Customer is unhappy with the quality.
119 * `too_complex`: Customer considers the service too complicated.
120 * `other`: Other reason(s).
121 """
122 ),
123 )
124 cancellation_comment: str | None = Field( 1a
125 None, description="Customer feedback and why they decided to cancel."
126 )
129CustomerSubscriptionUpdate = Annotated[ 1a
130 CustomerSubscriptionUpdateProduct
131 | CustomerSubscriptionUpdateSeats
132 | CustomerSubscriptionCancel,
133 SetSchemaReference("CustomerSubscriptionUpdate"),
134]