Coverage for polar/customer_portal/schemas/subscription.py: 97%

1import inspect 

2from typing import Annotated 

3 

4from pydantic import UUID4, AliasChoices, AliasPath, Field, computed_field 

5from pydantic.json_schema import SkipJsonSchema 

6 

7from polar.enums import SubscriptionProrationBehavior 

8from polar.kit.schemas import IDSchema, Schema, SetSchemaReference, TimestampedSchema 

9from polar.meter.schemas import NAME_DESCRIPTION as METER_NAME_DESCRIPTION 

10from polar.models.subscription import CustomerCancellationReason 

11from polar.product.schemas import ( 

12 BenefitPublicList, 

13 ProductBase, 

14 ProductMediaList, 

15 ProductPrice, 

16 ProductPriceList, 

17) 

18from polar.subscription.schemas import SubscriptionBase, SubscriptionMeterBase 

19 

20from .organization import CustomerOrganization 

21 

22 

23class CustomerSubscriptionProduct(ProductBase): 

24 prices: ProductPriceList 

25 benefits: BenefitPublicList 

26 medias: ProductMediaList 

27 organization: CustomerOrganization 

28 

29 

30class CustomerSubscriptionMeterMeter(IDSchema, TimestampedSchema): 

31 name: str = Field(description=METER_NAME_DESCRIPTION) 

32 

33 

34class CustomerSubscriptionMeter(SubscriptionMeterBase): 

35 meter: CustomerSubscriptionMeterMeter 

36 

37 

38class CustomerSubscription(SubscriptionBase): 

39 user_id: SkipJsonSchema[UUID4] = Field( 

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 

49 

50 price: SkipJsonSchema[ProductPrice] = Field( 

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 ) 

59 

60 prices: list[ProductPrice] = Field( 

61 description="List of enabled prices for the subscription." 

62 ) 

63 meters: list[CustomerSubscriptionMeter] = Field( 

64 description="List of meters associated with the subscription." 

65 ) 

66 

67 stripe_subscription_id: SkipJsonSchema[str | None] = Field( 

68 validation_alias="stripe_subscription_id" 

69 ) 

70 

71 @computed_field 

72 def is_polar_managed(self) -> bool: 

73 """Whether the subscription is managed by Polar.""" 

74 return self.stripe_subscription_id is None 

75 

76 

77class CustomerSubscriptionUpdateProduct(Schema): 

78 product_id: UUID4 = Field(description="Update subscription to another product.") 

79 

80 

81class CustomerSubscriptionUpdateSeats(Schema): 

82 seats: int = Field( 

83 description="Update the number of seats for this subscription.", 

84 ge=1, 

85 ) 

86 proration_behavior: SubscriptionProrationBehavior | None = Field( 

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 ) 

93 

94 

95class CustomerSubscriptionCancel(Schema): 

96 cancel_at_period_end: bool | None = Field( 

97 None, 

98 description=inspect.cleandoc( 

99 """ 

100 Cancel an active subscription once the current period ends. 

101 

102 Or uncancel a subscription currently set to be revoked at period end. 

103 """ 

104 ), 

105 ) 

106 

107 cancellation_reason: CustomerCancellationReason | None = Field( 

108 None, 

109 description=inspect.cleandoc( 

110 """ 

111 Customers reason for cancellation. 

112 

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( 

125 None, description="Customer feedback and why they decided to cancel." 

126 ) 

127 

128 

129CustomerSubscriptionUpdate = Annotated[ 

130 CustomerSubscriptionUpdateProduct 

131 | CustomerSubscriptionUpdateSeats 

132 | CustomerSubscriptionCancel, 

133 SetSchemaReference("CustomerSubscriptionUpdate"), 

134]